PythonとgRPCで実装する高性能マイクロサービス通信：プロダクションレベルガイド

# server.py
from concurrent import futures
import time
import grpc
import product_pb2
import product_pb2_grpc

# 仮想データベース
DUMMY_PRODUCTS = [
    product_pb2.Product(id="p001", name="Laptop Pro X", description="High-end laptop", price=1500.00, category="Electronics"),
    product_pb2.Product(id="p002", name="Wireless Mouse", description="Ergonomic mouse", price=75.50, category="Electronics"),
    product_pb2.Product(id="p003", name="Mechanical Keyboard", description="RGB Keyboard", price=120.00, category="Electronics"),
    product_pb2.Product(id="p004", name="The Python Guide", description="A book for Pythonistas", price=45.99, category="Books"),
]

class ProductServiceServicer(product_pb2_grpc.ProductServiceServicer):
    """ProductServiceの実際のロジックを実装するクラス"""

    def GetProduct(self, request, context):
        """Unary RPC: 商品IDで商品情報を照会"""
        print(f"Received GetProduct request for ID: {request.product_id}")
        for product in DUMMY_PRODUCTS:
            if product.id == request.product_id:
                return product
        
        # 商品が見つからない場合
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details(f"Product with ID '{request.product_id}' not found.")
        return product_pb2.Product()

    def ListProductsByCategory(self, request, context):
        """Server Streaming RPC: カテゴリで商品リストをストリームで返す"""
        print(f"Received ListProductsByCategory request for category: {request.category}")
        for product in DUMMY_PRODUCTS:
            if product.category == request.category:
                print(f"Streaming product: {product.name}")
                yield product
                time.sleep(1) # ストリーミングを視覚的に見せるための遅延

def serve():
    """gRPCサーバーを起動する関数"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    product_pb2_grpc.add_ProductServiceServicer_to_server(ProductServiceServicer(), server)
    
    server_address = '[::]:50051'
    server.add_insecure_port(server_address)
    
    print(f"🚀 Server starting on {server_address}")
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

# client.py
import grpc
import product_pb2
import product_pb2_grpc

def run():
    # サーバーとのチャネルを作成
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = product_pb2_grpc.ProductServiceStub(channel)

        # 1. GetProduct (Unary RPC) 呼び出し
        print("--- Calling GetProduct (p001) ---")
        try:
            product_response = stub.GetProduct(product_pb2.GetProductRequest(product_id="p001"))
            print(f"Product found: {product_response.name}, Price: ${product_response.price}")
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

        print("\n--- Calling GetProduct (p999 - not found) ---")
        try:
            stub.GetProduct(product_pb2.GetProductRequest(product_id="p999"))
        except grpc.RpcError as e:
            print(f"RPC failed as expected: {e.code()} - {e.details()}")

        # 2. ListProductsByCategory (Server Streaming RPC) 呼び出し
        print("\n--- Calling ListProductsByCategory (Electronics) ---")
        try:
            product_stream = stub.ListProductsByCategory(
                product_pb2.ListProductsByCategoryRequest(category="Electronics")
            )
            print("Receiving product stream...")
            for product in product_stream:
                print(f"  - Received: {product.name} (ID: {product.id})")
            print("Stream finished.")
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

if __name__ == '__main__':
    run()

# in Servicer class
def SomeRpcMethod(self, request, context):
    if not request.user_id:
        context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
        context.set_details("user_id field is required.")
        return some_pb2.SomeResponse()
    
    # ... logic ...

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        auth_token = metadata.get('authorization')

        if not self._is_valid_token(auth_token):
            # contextに直接アクセスできないため、エラーを発生させる特別なハンドラを返す
            return self._abort_with_status(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

        return continuation(handler_call_details)

    # ... helper methods ...

# client.py
try:
    # 5秒以内に応答がなければDEADLINE_EXCEEDEDエラーが発生
    response = stub.SlowRpcMethod(request, timeout=5) 
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("Request timed out!")