gRPC Services - Protocol Buffers, Proto Files, Services

Introduction

gRPC is a high-performance RPC framework that uses Protocol Buffers for serialization. It provides efficient communication between services with support for streaming and authentication.

Protocol Buffers Definition

// user.proto
syntax = "proto3";

package user;

service UserService {
    rpc GetUser (UserRequest) returns (User);
    rpc CreateUser (CreateUserRequest) returns (User);
    rpc ListUsers (ListUsersRequest) returns (UserList);
    rpc DeleteUser (UserRequest) returns (Empty);
    rpc StreamUsers (Empty) returns (stream User);
}

message User {
    string id = 1;
    string name = 2;
    string email = 3;
    string created_at = 4;
    repeated string roles = 5;
}

message UserRequest {
    string id = 1;
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
    repeated string roles = 3;
}

message ListUsersRequest {
    int32 page_size = 1;
    string page_token = 2;
}

message UserList {
    repeated User users = 1;
    string next_page_token = 2;
}

message Empty {}

Python gRPC Server

# server.py
import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserServiceServicer(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users = {}
    
    def GetUser(self, request, context):
        user = self.users.get(request.id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details('User not found')
            return user_pb2.User()
        return user_pb2.User(
            id=user['id'],
            name=user['name'],
            email=user['email'],
            created_at=user['created_at'],
            roles=user['roles']
        )
    
    def CreateUser(self, request, context):
        user_id = str(len(self.users) + 1)
        user = {
            'id': user_id,
            'name': request.name,
            'email': request.email,
            'created_at': '2024-01-01',
            'roles': list(request.roles)
        }
        self.users[user_id] = user
        return user_pb2.User(**user)
    
    def ListUsers(self, request, context):
        users = list(self.users.values())
        return user_pb2.UserList(
            users=[
                user_pb2.User(**u) for u in users[:request.page_size]
            ]
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

Python gRPC Client

# client.py
import grpc
import user_pb2
import user_pb2_grpc

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = user_pb2_grpc.UserServiceStub(channel)
    
    # Create user
    response = stub.CreateUser(user_pb2.CreateUserRequest(
        name="John Doe",
        email="john@example.com",
        roles=["admin", "user"]
    ))
    print(f"Created user: {response.id}")
    
    # Get user
    user = stub.GetUser(user_pb2.UserRequest(id=response.id))
    print(f"User: {user.name}, {user.email}")
    
    # List users
    users = stub.ListUsers(user_pb2.ListUsersRequest(page_size=10))
    print(f"Total users: {len(users.users)}")

if __name__ == '__main__':
    run()

Practice Problems

1. Create a proto file for a chat application with message types
2. Implement server-side streaming for sending notifications
3. Add authentication to gRPC service methods
4. Implement bidirectional streaming for real-time communication
5. Create a gRPC client with connection pooling