파이썬과 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!")