Premiers pas avec gRPC-Python : streaming

1. Introduction

Dans cet atelier de programmation, vous allez utiliser gRPC-Python pour créer un client et un serveur qui constituent la base d'une application de cartographie d'itinéraires écrite en Python.

À la fin du tutoriel, vous disposerez d'un client qui se connecte à un serveur distant à l'aide de gRPC pour obtenir des informations sur les caractéristiques d'un itinéraire client, créer un récapitulatif d'un itinéraire client et échanger des informations sur l'itinéraire, telles que les mises à jour du trafic, avec le serveur et d'autres clients.

Le service est défini dans un fichier Protocol Buffers, qui sera utilisé pour générer du code passe-partout pour le client et le serveur afin qu'ils puissent communiquer entre eux. Vous gagnerez ainsi du temps et des efforts pour implémenter cette fonctionnalité.

Ce code généré gère non seulement les complexités de la communication entre le serveur et le client, mais aussi la sérialisation et la désérialisation des données.

Points abordés

  • Utiliser Protocol Buffers pour définir une API de service.
  • Comment créer un client et un serveur basés sur gRPC à partir d'une définition Protocol Buffers à l'aide de la génération de code automatisée.
  • Comprendre la communication de streaming client-serveur avec gRPC.

Cet atelier de programmation s'adresse aux développeurs Python qui découvrent gRPC ou qui souhaitent se rafraîchir la mémoire sur gRPC, ou à toute personne intéressée par la création de systèmes distribués. Aucune expérience préalable avec gRPC n'est requise.

2. Avant de commencer

Prérequis

  • Python 3.9 ou version ultérieure. Nous vous recommandons Python 3.13. Pour obtenir des instructions d'installation spécifiques à la plate-forme, consultez Configuration et utilisation de Python. Vous pouvez également installer un Python non système à l'aide d'outils tels que uv ou pyenv.
  • pip pour installer les packages Python.
  • venv pour créer des environnements virtuels Python.

Les packages ensurepip et venv font partie de la bibliothèque standard Python et sont généralement disponibles par défaut.

Toutefois, certaines distributions basées sur Debian (y compris Ubuntu) choisissent de les exclure lors de la redistribution de Python. Pour installer les packages, exécutez la commande suivante :

sudo apt install python3-pip python3-venv

Obtenir le code

Pour faciliter votre apprentissage, cet atelier de programmation propose une structure de code source prédéfinie pour vous aider à démarrer. Les étapes suivantes vous guideront dans la création de l'application, y compris la génération de code gRPC à l'aide du plug-in du compilateur Protocol Buffer grpc_tools.protoc.

grpc-codelabs

Le code source du scaffold de cet atelier de programmation est disponible dans le répertoire codelabs/grpc-python-streaming/start_here. Si vous préférez ne pas implémenter le code vous-même, le code source complet est disponible dans le répertoire completed.

Commencez par créer le répertoire de travail de l'atelier de programmation et accédez-y :

mkdir grpc-python-streaming && cd grpc-python-streaming

Téléchargez et extrayez l'atelier de programmation :

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-1/codelabs/grpc-python-streaming/start_here

Vous pouvez également télécharger le fichier .zip contenant uniquement le répertoire de l'atelier de programmation et le décompresser manuellement.

3. Définir les messages et les services

La première étape consiste à définir le service gRPC de l'application, sa méthode RPC, ainsi que ses types de messages de requête et de réponse à l'aide de Protocol Buffers. Votre service fournira :

  • Méthodes RPC appelées ListFeatures, RecordRoute et RouteChat que le serveur implémente et que le client appelle.
  • Les types de messages Point, Feature, Rectangle, RouteNote et RouteSummary, qui sont des structures de données échangées entre le client et le serveur lors de l'appel des méthodes RPC.

Ces méthodes RPC et leurs types de messages seront tous définis dans le fichier protos/route_guide.proto du code source fourni.

Les tampons de protocole sont communément appelés "protobufs". Pour en savoir plus sur la terminologie gRPC, consultez Concepts fondamentaux, architecture et cycle de vie de gRPC.

Définir les types de messages

Dans le fichier protos/route_guide.proto du code source, définissez d'abord le type de message Point. Un Point représente une paire de coordonnées (latitude et longitude) sur une carte. Pour cet atelier de programmation, utilisez des nombres entiers pour les coordonnées :

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

Les nombres 1 et 2 sont des ID uniques pour chacun des champs de la structure message.

Ensuite, définissez le type de message Feature. Un Feature utilise un champ string pour le nom ou l'adresse postale d'un élément à un emplacement spécifié par un Point :

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

Pour que plusieurs points d'une zone puissent être diffusés en streaming vers un client, vous avez besoin d'un message Rectangle qui représente un rectangle de latitude et de longitude, représenté par deux points diagonalement opposés lo et hi :

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

Un message RouteNote qui représente un message envoyé à un moment donné :

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

Enfin, vous aurez besoin d'un message RouteSummary. Ce message est reçu en réponse à un RPC RecordRoute, qui est expliqué dans la section suivante. Il contient le nombre de points individuels reçus, le nombre de caractéristiques détectées et la distance totale parcourue, qui correspond à la somme cumulée de la distance entre chaque point.

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

Définir les méthodes de service

Pour définir un service, vous devez spécifier un service nommé dans votre fichier .proto. Le fichier route_guide.proto possède une structure service nommée RouteGuide qui définit une ou plusieurs méthodes fournies par le service de l'application.

Lorsque vous définissez des méthodes RPC dans la définition de votre service, vous spécifiez leurs types de requête et de réponse. Dans cette section de l'atelier de programmation, définissons les éléments suivants :

ListFeatures

Obtient les objets Feature disponibles dans le Rectangle donné. Les résultats sont diffusés en flux continu plutôt que renvoyés en une seule fois, car le rectangle peut couvrir une grande zone et contenir un grand nombre d'entités.

Pour cette application, vous utiliserez un RPC de streaming côté serveur : le client envoie une requête au serveur et obtient un flux pour lire une séquence de messages en retour. Le client lit le flux renvoyé jusqu'à ce qu'il n'y ait plus de messages. Comme vous pouvez le voir dans notre exemple, vous spécifiez une méthode de streaming côté serveur en plaçant le mot clé "stream" avant le type de réponse.

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

Accepte un flux de points sur un itinéraire parcouru et renvoie un RouteSummary lorsque le parcours est terminé.

Un RPC de streaming côté client est approprié dans ce cas : le client écrit une séquence de messages et les envoie au serveur, à nouveau à l'aide d'un flux fourni. Une fois que le client a terminé d'écrire les messages, il attend que le serveur les lise tous et renvoie sa réponse. Pour spécifier une méthode de streaming côté client, placez le mot clé "stream" avant le type de requête.

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

Accepte un flux de RouteNotes envoyé lors du parcours d'un itinéraire, tout en recevant d'autres RouteNotes (par exemple, d'autres utilisateurs).

C'est exactement le cas d'utilisation du streaming bidirectionnel. Un RPC de streaming bidirectionnel où les deux côtés envoient une séquence de messages à l'aide d'un flux en lecture-écriture. Les deux flux fonctionnent indépendamment. Les clients et les serveurs peuvent donc lire et écrire dans l'ordre de leur choix. Par exemple, le serveur peut attendre de recevoir tous les messages du client avant d'écrire ses réponses, ou il peut lire un message puis en écrire un, ou encore effectuer une autre combinaison de lectures et d'écritures. L'ordre des messages dans chaque flux est conservé. Pour spécifier ce type de méthode, placez le mot clé "stream" avant la requête et la réponse.

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. Générer le code client et serveur

Ensuite, générez le code gRPC standard pour le client et le serveur à partir du fichier .proto à l'aide du compilateur de tampons de protocole.

Pour la génération de code Python gRPC, nous avons créé grpcio-tools. Il inclut les éléments suivants :

  1. Le compilateur protoc standard qui génère du code Python à partir des définitions message.
  2. Plug-in protobuf gRPC qui génère du code Python (stubs client et serveur) à partir des définitions service.

Nous allons installer le package Python grpcio-tools à l'aide de pip. Créons un environnement virtuel Python (venv) pour isoler les dépendances de votre projet des packages système :

python3 -m venv --upgrade-deps .venv

Pour activer l'environnement virtuel dans le shell bash/zsh :

source .venv/bin/activate

Pour Windows et les shells non standards, consultez le tableau à l'adresse https://docs.python.org/3/library/venv.html#how-venvs-work.

Ensuite, installez grpcio-tools (cela installe également le package grpcio) :

pip install grpcio-tools

Utilisez la commande suivante pour générer le code standard Python :

python -m grpc_tools.protoc --proto_path=./protos  \
 --python_out=. --pyi_out=. --grpc_python_out=. \
 ./protos/route_guide.proto

Cette opération génère les fichiers suivants pour les interfaces que nous avons définies dans route_guide.proto :

  1. route_guide_pb2.py contient le code qui crée dynamiquement des classes générées à partir des définitions message.
  2. route_guide_pb2.pyi est un fichier stub ou un fichier d'indication de type généré à partir des définitions message. Il ne contient que les signatures sans implémentation. Les fichiers de stub peuvent être utilisés par les IDE pour améliorer l'autocomplétion et la détection des erreurs.
  3. route_guide_pb2_grpc.py est généré à partir des définitions service et contient des classes et des fonctions spécifiques à gRPC.

Le code spécifique à gRPC contient :

  1. RouteGuideStub, qui peut être utilisé par un client gRPC pour appeler les RPC RouteGuide.
  2. RouteGuideServicer, qui définit l'interface pour les implémentations du service RouteGuide.
  3. La fonction add_RouteGuideServicer_to_server est utilisée pour enregistrer un RouteGuideServicer sur un serveur gRPC.

5. Créer le serveur

Commençons par voir comment créer un serveur RouteGuide. La création et l'exécution d'un serveur RouteGuide se décomposent en deux tâches :

  • Implémenter l'interface de service générée à partir de notre définition de service avec des fonctions qui effectuent le "travail" réel du service.
  • Exécuter un serveur gRPC pour écouter les requêtes des clients et transmettre les réponses.

Examinons route_guide_server.py.

Implémenter RouteGuide

route_guide_server.py comporte une classe RouteGuideServicer qui est une sous-classe de la classe générée route_guide_pb2_grpc.RouteGuideServicer :

# RouteGuideServicer provides an implementation of the methods of the RouteGuide service.
class RouteGuideServicer(route_guide_pb2_grpc.RouteGuideServicer):

RouteGuideServicer implémente toutes les méthodes de service RouteGuide.

RPC de streaming côté serveur

ListFeatures est un RPC de streaming de réponse qui envoie plusieurs Feature au client :

def ListFeatures(self, request, context):
    """List all features contained within the given Rectangle."""
    left = min(request.lo.longitude, request.hi.longitude)
    right = max(request.lo.longitude, request.hi.longitude)
    top = max(request.lo.latitude, request.hi.latitude)
    bottom = min(request.lo.latitude, request.hi.latitude)
    for feature in self.db:
        lat, lng = feature.location.latitude, feature.location.longitude
        if left <= lng <= right and bottom <= lat <= top:
            yield feature

Ici, le message de requête est un route_guide_pb2.Rectangle dans lequel le client souhaite trouver des Feature. Au lieu de renvoyer une seule réponse, la méthode génère zéro ou plusieurs réponses.

RPC de streaming côté client

La méthode de streaming des requêtes RecordRoute utilise un itérateur de valeurs de requête et renvoie une seule valeur de réponse.

def RecordRoute(self, request_iterator, context):
    """Calculate statistics about the trip composed of Points."""
    point_count = 0
    feature_count = 0
    distance = 0.0
    prev_point = None

    start_time = time.time()
    for point in request_iterator:
        point_count += 1
        if get_feature(self.db, point):
            feature_count += 1
        if prev_point:
            distance += get_distance(prev_point, point)
        prev_point = point

    elapsed_time = time.time() - start_time
    return route_guide_pb2.RouteSummary(
        point_count=point_count,
        feature_count=feature_count,
        distance=int(distance),
        elapsed_time=int(elapsed_time),
    )

RPC de streaming bidirectionnel

Enfin, examinons notre RPC de streaming bidirectionnel RouteChat() :

def RouteChat(self, request_iterator, context):
    """
    Receive a stream of message/location pairs, and responds with
    a stream of all previous messages for the given location.
    """
    prev_notes = []
    for new_note in request_iterator:
        for prev_note in prev_notes:
            if prev_note.location == new_note.location:
                yield prev_note
        prev_notes.append(new_note)

La sémantique de cette méthode est une combinaison de celles de la méthode de streaming des requêtes et de la méthode de streaming des réponses. Il reçoit un itérateur de valeurs de requête et est lui-même un itérateur de valeurs de réponse.

Démarrer le serveur

Une fois que vous avez implémenté toutes les méthodes RouteGuide, l'étape suivante consiste à démarrer un serveur gRPC afin que les clients puissent réellement utiliser votre service :

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    route_guide_pb2_grpc.add_RouteGuideServicer_to_server(
        RouteGuideServicer(),
        server,
    )
    listen_addr = "localhost:50051"
    server.add_insecure_port(listen_addr)
    print(f"Starting server on {listen_addr}")
    server.start()
    server.wait_for_termination()

La méthode start() du serveur n'est pas bloquante. Un nouveau thread sera instancié pour gérer les requêtes. Le thread appelant server.start() n'aura souvent aucune autre tâche à effectuer en attendant. Dans ce cas, vous pouvez appeler server.wait_for_termination() pour bloquer proprement le thread appelant jusqu'à ce que le serveur se termine.

6. Créer le client

Examinons route_guide_client.py.

Créer un stub

Pour appeler des méthodes de service, nous devons d'abord créer un stub.

Nous instancions la classe RouteGuideStub du module route_guide_pb2_grpc, générée à partir de notre méthode .proto. dans run() :

with grpc.insecure_channel("localhost:50051") as channel:
    stub = route_guide_pb2_grpc.RouteGuideStub(channel)

Notez que channel est utilisé ici comme gestionnaire de contexte et sera automatiquement fermé une fois que l'interpréteur aura quitté le bloc with.

Appeler les méthodes de service

Pour les méthodes RPC qui renvoient une seule réponse (méthodes "response-unary"), gRPC Python est compatible avec les sémantiques de flux de contrôle synchrone (bloquant) et asynchrone (non bloquant). Pour les méthodes RPC de streaming de réponse, les appels renvoient immédiatement un itérateur de valeurs de réponse. Les appels à la méthode next() de cet itérateur sont bloqués jusqu'à ce que la réponse à générer à partir de l'itérateur devienne disponible.

RPC de streaming côté serveur

L'appel de ListFeatures pour le streaming des réponses est semblable à l'utilisation des types de séquence :

def guide_list_features(stub):
    _lo = route_guide_pb2.Point(latitude=400000000, longitude=-750000000)
    _hi = route_guide_pb2.Point(latitude=420000000, longitude=-730000000)
    rectangle = route_guide_pb2.Rectangle(
        lo=_lo,
        hi=_hi,
    )
    print("Looking for features between 40, -75 and 42, -73")

    features = stub.ListFeatures(rectangle)
    for feature in features:
        print(
            f"Feature called '{feature.name}'"
            f" at {format_point(feature.location)}"
        )

RPC de streaming côté client

Appeler le RecordRoute de streaming de requête revient à transmettre un itérateur à une méthode locale. Comme le RPC simple ci-dessus qui renvoie également une seule réponse, il peut être appelé de manière synchrone :

def guide_record_route(stub):
    feature_list = route_guide_resources.read_route_guide_database()
    route_iterator = generate_route(feature_list)

    route_summary = stub.RecordRoute(route_iterator)
    print(f"Finished trip with {route_summary.point_count} points")
    print(f"Passed {route_summary.feature_count} features")
    print(f"Traveled {route_summary.distance} meters")
    print(f"It took {route_summary.elapsed_time} seconds")

RPC de streaming bidirectionnel

L'appel de RouteChat en streaming bidirectionnel présente (comme c'est le cas côté service) une combinaison des sémantiques de streaming des requêtes et des réponses.

Générez les messages de requête et envoyez-les un par un à l'aide de yield.

def generate_notes():
    home = route_guide_pb2.Point(latitude=1, longitude=1)
    work = route_guide_pb2.Point(latitude=2, longitude=2)
    notes = [
        make_route_note("Departing from home", home),
        make_route_note("Arrived at work", work),
        make_route_note("Having lunch at work", work),
        make_route_note("Departing from work", work),
        make_route_note("Arrived home", home),
    ]
    for note in notes:
        print(
            f"Sending RouteNote for {format_point(note.location)}:"
            f" {note.message}"
        )
        yield note
        # Sleep to simulate moving from one point to another.
        # Only for demonstrating the order of the messages.
        time.sleep(0.1)

Recevez et traitez les réponses du serveur :

def guide_route_chat(stub):
    responses = stub.RouteChat(generate_notes())
    for response in responses:
        print(
            "< Found previous note at"
            f" {format_point(response.location)}: {response.message}"
        )

Appeler les méthodes d'assistance

Dans l'exécution, exécutez les méthodes que nous venons de créer et transmettez-leur le stub.

print("-------------- ListFeatures --------------")
guide_list_features(stub)
print("-------------- RecordRoute --------------")
guide_record_route(stub)
print("-------------- RouteChat --------------")
guide_route_chat(stub)

7. Essayer

Exécutez le serveur :

python route_guide_server.py

Depuis un autre terminal, activez à nouveau l'environnement virtuel (source .venv/bin/activate)), puis exécutez le client :

python route_guide_client.py

Examinons le résultat.

ListFeatures

Vous trouverez d'abord la liste des fonctionnalités. Chaque entité est diffusée en streaming depuis le serveur (RPC de streaming côté serveur) à mesure qu'elle est découverte dans le rectangle demandé :

-------------- ListFeatures --------------
Looking for features between 40, -75 and 42, -73
Feature called 'Patriots Path, Mendham, NJ 07945, USA' at (lat=407838351, lng=-746143763)
Feature called '101 New Jersey 10, Whippany, NJ 07981, USA' at (lat=408122808, lng=-743999179)
Feature called 'U.S. 6, Shohola, PA 18458, USA' at (lat=413628156, lng=-749015468)
Feature called '5 Conners Road, Kingston, NY 12401, USA' at (lat=419999544, lng=-740371136)
...

RecordRoute

Ensuite, RecordRoute montre la liste des points visités de manière aléatoire, diffusée du client vers le serveur (RPC de streaming côté client) :

-------------- RecordRoute --------------
Visiting point (lat=410395868, lng=-744972325)
Visiting point (lat=404310607, lng=-740282632)
Visiting point (lat=403966326, lng=-748519297)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=406589790, lng=-743560121)
Visiting point (lat=410322033, lng=-747871659)
Visiting point (lat=415464475, lng=-747175374)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=402647019, lng=-747071791)
Visiting point (lat=414638017, lng=-745957854)

Une fois que le client a terminé de diffuser tous les points visités, il reçoit une réponse sans streaming (un RPC unaire) du serveur. Cette réponse contient un récapitulatif des calculs effectués sur l'itinéraire complet du client.

Finished trip with 10 points
Passed 10 features
Traveled 654743 meters
It took 0 seconds

RouteChat

Enfin, la sortie RouteChat illustre le streaming bidirectionnel. Lorsque le client "visite" les points home ou work, il enregistre une note pour le point en envoyant un RouteNote au serveur. Lorsqu'un point a déjà été visité, le serveur renvoie en flux continu toutes les notes précédentes pour ce point.

-------------- RouteChat --------------
Sending RouteNote for (lat=1, lng=1): Departing from home
Sending RouteNote for (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Having lunch at work
< Found previous note at (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Departing from work
< Found previous note at (lat=2, lng=2): Arrived at work
< Found previous note at (lat=2, lng=2): Having lunch at work
Sending RouteNote for (lat=1, lng=1): Arrived home
< Found previous note at (lat=1, lng=1): Departing from home

8. Étape suivante