Client-Streaming RPC: Sending Data Streams

The Client-Streaming RPC Model

Let's visualize the four fundamental gRPC communication patterns. We have already covered Unary and Server Streaming. Today, we focus on the third pattern: Client Streaming.

A classic real-world example of client streaming is a file upload. The client reads a large file in chunks and streams these chunks to the server, which reassembles them. Once the final chunk is received, the server can send back a single confirmation response.

Step 1: Updating the Service Definition

As always, our first step is to define the new RPC in our user_service.proto file. To define a client-streaming RPC, you place the stream keyword before the request type in your service definition.

The article "How to Build gRPC Clients in Python" provides a clear example of this.

Let's apply this to our myservice/generated/user_service.proto. We will add a ListUsersResponse message and the BatchCreateUsers RPC to our UserService.

Update user_service.proto:

// ... (keep the existing content: syntax, package, User, UserStatus, CreateUserRequest, CreateUserResponse, GetUsersRequest)

// Add the new response message for batch operations
message ListUsersResponse {
  repeated User users = 1;
}

service UserService {
  // Unary RPC
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);

  // Server-streaming RPC
  rpc GetUsers(GetUsersRequest) returns (stream User);

  // Client-streaming RPC
  rpc BatchCreateUsers(stream CreateUserRequest) returns (ListUsersResponse);
}

After saving the updated .proto file, you must regenerate the Python stubs. From your project's root directory, run the protoc command again:

python -m grpc_tools.protoc -I=myservice/generated --python_out=myservice/generated --pyi_out=myservice/generated --grpc_python_out=myservice/generated myservice/generated/user_service.proto

This updates the generated Python files with the new message type and the necessary code for our client-streaming RPC.

Step 2: Implementing the Server-Side Stream Consumer

On the server, implementing a client-streaming RPC is the inverse of a server-streaming one. Instead of yielding responses, your method will receive an asynchronous iterator as its request argument. You'll use an async for loop to process each message as it arrives from the client. After the client signals that it has finished sending messages (closing its side of the stream), your method can perform any final processing and then return a single response.

The article "How to Handle Streaming in gRPC Services" demonstrates this server-side pattern clearly.

Let's now modify our myservice/server.py to implement the BatchCreateUsers method.

Update server.py:

# myservice/server.py

import asyncio
import logging
import uuid

import grpc
from generated import user_service_pb2
from generated import user_service_pb2_grpc

_USERS = [



    # ... (existing users)
]

class UserService(user_service_pb2_grpc.UserServiceServicer):
    """
    Provides the implementation for the UserService's RPCs.
    """
    async def CreateUser(



        # ... (no changes to this method)
    ) -> user_service_pb2.CreateUserResponse:



        # ...
    
    async def GetUsers(



        # ... (no changes to this method)
    ) -> user_service_pb2.User:



        # ...




    # Add the new client-streaming method
    async def BatchCreateUsers(
        self,
        request_iterator: grpc.aio.AsyncIterator[user_service_pb2.CreateUserRequest],
        context: grpc.aio.ServicerContext,
    ) -> user_service_pb2.ListUsersResponse:
        """
        Implements the Client-Streaming RPC for batch-creating users.
        """
        logging.info("Received BatchCreateUsers request. Awaiting user stream...")
        created_users = []
        



        # Use 'async for' to iterate over the stream of requests from the client.
        async for request in request_iterator:
            logging.info(f"Processing user from stream: {request.email}")
            user = user_service_pb2.User(
                id=str(uuid.uuid4()),
                name=request.name,
                email=request.email,
                age=request.age,
                roles=request.roles,
                status=user_service_pb2.UserStatus.ACTIVE
            )
            _USERS.append(user)
            created_users.append(user)

        logging.info(f"Finished processing user stream. Created {len(created_users)} users.")
        



        # Return a single response containing all the created users.
        return user_service_pb2.ListUsersResponse(users=created_users)

async def serve() -> None:



    # ... (no changes to this function)

if __name__ == "__main__":



    # ... (no changes here)

Key changes:

• The BatchCreateUsers method receives request_iterator, an async iterator.
• We use an async for loop to process each CreateUserRequest as it arrives.
• Inside the loop, we create a new User and add it to both our in-memory _USERS list and a temporary created_users list.
• After the loop completes, we return a single ListUsersResponse that aggregates the results.

Step 3: Implementing the Client-Side Stream Producer

On the client side, we need to provide a stream of requests. With asyncio, the idiomatic way to do this is by creating an asynchronous generator function. This function will yield each request object. You then pass this generator function to the gRPC stub method.

The stub method call itself is awaited, because it will eventually resolve to the single response message returned by the server once the stream is fully processed.

The "How to Build gRPC Clients in Python" article provides an excellent async client implementation that we can follow.

Let's modify myservice/client.py to call our new BatchCreateUsers endpoint.

Update client.py:

# myservice/client.py

import asyncio
import logging

import grpc
from generated import user_service_pb2
from generated import user_service_pb2_grpc

async def call_create_user(stub: user_service_pb2_grpc.UserServiceStub):



    # ... (no changes)

async def call_get_users(stub: user_service_pb2_grpc.UserServiceStub):



    # ... (no changes)

async def call_batch_create_users(stub: user_service_pb2_grpc.UserServiceStub):
    """Calls the Client-Streaming BatchCreateUsers RPC."""
    logging.info("--- Client will call BatchCreateUsers ---")
    



    # Data for the users we want to create in a batch.
    new_users_data = [
        {"name": "Kevin", "email": "kevin@example.com", "age": 40, "roles": ["admin"]},
        {"name": "Laura", "email": "laura@example.com", "age": 32, "roles": ["user"]},
        {"name": "Mike", "email": "mike@example.com", "age": 25, "roles": ["user"]},
    ]




    # This is our async generator that produces the stream of requests.
    async def request_generator():
        for user_data in new_users_data:
            logging.info(f"Client streaming user: {user_data['name']}")
            yield user_service_pb2.CreateUserRequest(
                name=user_data["name"],
                email=user_data["email"],
                age=user_data["age"],
                roles=user_data["roles"],
            )



            # Simulate some work or delay between sending requests
            await asyncio.sleep(0.5)

    try:



        # We pass the generator to the stub method and await the final response.
        response = await stub.BatchCreateUsers(request_generator(), timeout=30)
        logging.info(f"Client received response from BatchCreateUsers.")
        logging.info(f"Total users created: {len(response.users)}")
        for user in response.users:
            logging.info(f" - Created user: {user.name} with ID: {user.id}")

    except grpc.aio.AioRpcError as e:
        logging.error(f"BatchCreateUsers RPC failed: {e.code()} - {e.details()}")


async def run() -> None:
    """Creates a client and makes RPC calls to the server."""
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        
        await call_create_user(stub)
        print("-" * 20)
        await call_get_users(stub)
        print("-" * 20)
        await call_batch_create_users(stub)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run())

Key changes:

• We created a new function call_batch_create_users.
• Inside it, we defined an async generator request_generator that yields CreateUserRequest messages.
• The generator request_generator() is passed directly to stub.BatchCreateUsers. The gRPC library handles iterating over it and sending the messages.
• We await the result of stub.BatchCreateUsers, as it will eventually return a single ListUsersResponse.

Step 4: Running the Application

The execution process remains the same. Use two separate terminals.

Terminal 1: Start the Server
In your myservice/ directory, run:

python server.py

The server will start and wait for connections.

Terminal 2: Run the Client
In another terminal, also in the myservice/ directory, run:

python client.py

Observe the output in both terminals. The client will first call the unary and server-streaming RPCs as before. Then, for the client-streaming call, you'll see the client log each user it's streaming. Simultaneously, the server will log each user it receives. Finally, after a short delay, the client will log the single summary response from the server.

Client Output:

... (output from previous RPCs) ...
--------------------
INFO:root:--- Client will call BatchCreateUsers ---
INFO:root:Client streaming user: Kevin
INFO:root:Client streaming user: Laura
INFO:root:Client streaming user: Mike
INFO:root:Client received response from BatchCreateUsers.
INFO:root:Total users created: 3
INFO:root: - Created user: Kevin with ID: ...
INFO:root: - Created user: Laura with ID: ...
INFO:root: - Created user: Mike with ID: ...

Server Output:

... (output from previous RPCs) ...
INFO:root:Received BatchCreateUsers request. Awaiting user stream...
INFO:root:Processing user from stream: kevin@example.com
INFO:root:Processing user from stream: laura@example.com
INFO:root:Processing user from stream: mike@example.com
INFO:root:Finished processing user stream. Created 3 users.

This interaction perfectly demonstrates the client-streaming pattern: the client sends multiple messages, the server processes them, and then the client receives a single, final response.

Conclusion

Congratulations! You have now mastered client-streaming RPCs in async Python. This pattern is essential for building efficient data pipelines and handling large-scale data ingestion in your microservices.

Key Takeaways:

• Service Definition: A client-streaming RPC is defined in the .proto file by adding the stream keyword to the request type.
• Server Implementation: The server method receives an async iterator for the request. It uses an async for loop to process the incoming stream and returns a single response after the stream is exhausted.
• Client Implementation: The client creates an async generator to yield request messages. This generator is passed to the stub method, and the client awaits the single, final response from the server.

Preview of the Next Lesson

You have now seen data stream from server to client, and from client to server. What if you need both at the same time? In our next lesson, we will tackle the final and most powerful gRPC communication pattern: bidirectional-streaming RPC. You'll learn how a client and server can send independent streams of messages to each other over a single connection, enabling real-time, interactive applications like chat services or live collaboration tools.