gRPC & Protocol Buffers Proficient¶

🌐 Networking · Level 4

⏱️ ~4 days 📚 Prerequisites: Sockets, APIs & HTTP

When you'd use this

High-performance RPC with protobuf, streaming and service definitions.

Define typed service contracts with Protocol Buffers for fast, cross-language service-to-service communication.

What is gRPC?¶

Introduces gRPC and where it fits in gRPC & Protocol Buffers.

gRPC is a high-performance RPC framework using Protocol Buffers for serialization — 10x faster than JSON REST for inter-service communication.

Define service with .proto¶

Declare messages and RPC methods in a language-neutral schema that generates typed stubs.

// user_service.proto
syntax = "proto3";

package userservice;

service UserService {
    rpc GetUser (GetUserRequest) returns (User);
    rpc ListUsers (ListUsersRequest) returns (stream User);  // server streaming
    rpc CreateUser (User) returns (User);
}

message User {
    int32 id = 1;
    string name = 2;
    string email = 3;
    int32 age = 4;
}

message GetUserRequest {
    int32 id = 1;
}

message ListUsersRequest {
    int32 page_size = 1;
}

# Generate Python code
pip install grpcio grpcio-tools
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. user_service.proto

Server implementation¶

Implement the generated service interface to handle RPC calls.

import grpc
from concurrent import futures
import user_service_pb2 as pb2
import user_service_pb2_grpc as pb2_grpc

class UserServicer(pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users = {
            1: pb2.User(id=1, name="Alice", email="alice@example.com", age=30),
            2: pb2.User(id=2, name="Bob", email="bob@example.com", age=25),
        }

    def GetUser(self, request, context):
        user = self.users.get(request.id)
        if not user:
            context.abort(grpc.StatusCode.NOT_FOUND, f"User {request.id} not found")
        return user

    def ListUsers(self, request, context):
        """Server streaming — yields users one by one."""
        for user in self.users.values():
            yield user

    def CreateUser(self, request, context):
        new_id = max(self.users.keys()) + 1
        user = pb2.User(id=new_id, name=request.name, email=request.email, age=request.age)
        self.users[new_id] = user
        return user

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    pb2_grpc.add_UserServiceServicer_to_server(UserServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print("gRPC server on :50051")
    server.wait_for_termination()

serve()

Client¶

Call remote methods as if they were local functions using the generated stub.

import grpc
import user_service_pb2 as pb2
import user_service_pb2_grpc as pb2_grpc

channel = grpc.insecure_channel("localhost:50051")
stub = pb2_grpc.UserServiceStub(channel)

# Unary call
user = stub.GetUser(pb2.GetUserRequest(id=1))
print(f"Got: {user.name} ({user.email})")   # Alice (alice@example.com)

# Server streaming
for user in stub.ListUsers(pb2.ListUsersRequest(page_size=10)):
    print(f"  {user.id}: {user.name}")

# Create
new_user = stub.CreateUser(pb2.User(name="Charlie", email="c@d.com", age=35))
print(f"Created: {new_user.id}")

Practice Exercises¶

Define a proto for an order service with CRUD operations.
Implement server streaming — stream real-time price updates to clients.
Add error handling — return proper gRPC status codes for validation errors.
Benchmark gRPC vs REST for 1000 requests — compare latency and throughput.
Add authentication with gRPC interceptors (like middleware).

💬 Discussion

Have a question about this topic? Found an error? Share your thoughts below.