Python gRPC 개념 정리

1️⃣ gRPC 기초

gRPC는 구글이 개발한 고성능 RPC(Remote Procedure Call) 프레임워크이다.

# 프로토버프 정의 (user.proto)
syntax = "proto3";

package user;

// 서비스 정의
service UserService {
    // 단일 요청/응답 메서드
    rpc GetUser (UserRequest) returns (UserResponse) {}
    // 사용자 생성 메서드
    rpc CreateUser (CreateUserRequest) returns (UserResponse) {}
    // 사용자 목록 조회 메서드
    rpc ListUsers (ListUsersRequest) returns (ListUsersResponse) {}
    // 사용자 업데이트 메서드
    rpc UpdateUser (UpdateUserRequest) returns (UserResponse) {}
    // 사용자 삭제 메서드
    rpc DeleteUser (UserRequest) returns (DeleteUserResponse) {}
}

// 요청/응답 메시지 정의
message UserRequest {
    string user_id = 1;
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
    string phone = 3;
    int32 age = 4;
}

message UpdateUserRequest {
    string user_id = 1;
    string name = 2;
    string email = 3;
    string phone = 4;
    int32 age = 5;
}

message ListUsersRequest {
    int32 page_size = 1;  // 페이지당 항목 수
    int32 page_number = 2;  // 페이지 번호
    string sort_by = 3;  // 정렬 기준 필드
}

message UserResponse {
    string user_id = 1;
    string name = 2;
    string email = 3;
    string phone = 4;
    int32 age = 5;
    string created_at = 6;
    string updated_at = 7;
}

message ListUsersResponse {
    repeated UserResponse users = 1;
    int32 total_count = 2;
    int32 total_pages = 3;
}

message DeleteUserResponse {
    bool success = 1;
    string message = 2;
}

✅ 특징:

프로토콜 버퍼
강력한 타입 시스템
다중 언어 지원
높은 성능 및 효율성
코드 생성 자동화
양방향 스트리밍 지원

gRPC vs REST 비교:

gRPC와 REST는 API 설계에 있어 각각 다른 장단점을 가지고 있다.

기능	gRPC	REST
프로토콜	HTTP/2	HTTP/1.1 또는 HTTP/2
데이터 형식	프로토콜 버퍼 (이진)	JSON, XML (텍스트)
코드 생성	자동 생성	수동 또는 도구 사용
스트리밍	양방향 스트리밍	제한적
타입 안전성	강력한 타입 검사	스키마 기반 검증 필요
브라우저 지원	제한적	네이티브 지원
성능	효율적인 데이터 전송	상대적으로 오버헤드 큼
학습 곡선	가파름	완만함

2️⃣ 서버 구현

Python에서 gRPC 서버를 구현하는 방법과 서비스 로직 작성 방법이다.

import grpc
from concurrent import futures
import time
import logging
import user_pb2
import user_pb2_grpc
from database import UserDatabase

# 로깅 설정
logging.basicConfig(level=logging.INFO)

# 서비스 구현 클래스
class UserServicer(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        # 데이터베이스 연결 (예시)
        self.db = UserDatabase()
        
    def GetUser(self, request, context):
        logging.info(f"GetUser 요청 수신: {request.user_id}")
        try:
            # 사용자 조회 (가상 함수)
            user = self.db.find_user(request.user_id)
            if not user:
                context.set_code(grpc.StatusCode.NOT_FOUND)
                context.set_details(f"사용자 ID {request.user_id}를 찾을 수 없습니다")
                return user_pb2.UserResponse()
            
            # 응답 생성
            return user_pb2.UserResponse(
                user_id=user.id,
                name=user.name,
                email=user.email,
                phone=user.phone,
                age=user.age,
                created_at=user.created_at.isoformat(),
                updated_at=user.updated_at.isoformat()
            )
        except Exception as e:
            logging.error(f"GetUser 오류: {str(e)}")
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f"내부 서버 오류: {str(e)}")
            return user_pb2.UserResponse()
    
    def CreateUser(self, request, context):
        logging.info(f"CreateUser 요청 수신: {request.name}, {request.email}")
        try:
            # 입력 검증
            if not request.name or not request.email:
                context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
                context.set_details("이름과 이메일은 필수 항목입니다")
                return user_pb2.UserResponse()
            
            # 이메일 중복 확인
            if self.db.email_exists(request.email):
                context.set_code(grpc.StatusCode.ALREADY_EXISTS)
                context.set_details(f"이메일 {request.email}는 이미 사용 중입니다")
                return user_pb2.UserResponse()
            
            # 사용자 생성 (가상 함수)
            user = self.db.create_user(
                name=request.name,
                email=request.email,
                phone=request.phone,
                age=request.age
            )
            
            # 응답 생성
            return user_pb2.UserResponse(
                user_id=user.id,
                name=user.name,
                email=user.email,
                phone=user.phone,
                age=user.age,
                created_at=user.created_at.isoformat(),
                updated_at=user.updated_at.isoformat()
            )
        except Exception as e:
            logging.error(f"CreateUser 오류: {str(e)}")
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f"내부 서버 오류: {str(e)}")
            return user_pb2.UserResponse()
    
    def ListUsers(self, request, context):
        logging.info(f"ListUsers 요청 수신: 페이지 {request.page_number}, 크기 {request.page_size}")
        try:
            # 페이지네이션 처리 (가상 함수)
            users, total = self.db.list_users(
                page_size=request.page_size,
                page_number=request.page_number,
                sort_by=request.sort_by
            )
            
            # 사용자 목록 구성
            user_responses = []
            for user in users:
                user_responses.append(user_pb2.UserResponse(
                    user_id=user.id,
                    name=user.name,
                    email=user.email,
                    phone=user.phone,
                    age=user.age,
                    created_at=user.created_at.isoformat(),
                    updated_at=user.updated_at.isoformat()
                ))
            
            # 총 페이지 수 계산
            total_pages = (total + request.page_size - 1) // request.page_size
            
            # 응답 생성
            return user_pb2.ListUsersResponse(
                users=user_responses,
                total_count=total,
                total_pages=total_pages
            )
        except Exception as e:
            logging.error(f"ListUsers 오류: {str(e)}")
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f"내부 서버 오류: {str(e)}")
            return user_pb2.ListUsersResponse()

# 서버 시작 함수
def serve():
    # 서버 생성
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    
    # 서비스 등록
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServicer(), server
    )
    
    # 포트 바인딩
    server.add_insecure_port('[::]:50051')
    
    # 서버 시작
    server.start()
    logging.info("서버가 포트 50051에서 시작되었습니다")
    
    try:
        # 서버 실행 유지
        while True:
            time.sleep(86400)  # 1일
    except KeyboardInterrupt:
        # 서버 종료
        server.stop(0)
        logging.info("서버가 종료되었습니다")

if __name__ == '__main__':
    serve()

✅ 특징:

서비스 구현
요청 처리
응답 생성
오류 처리
비동기 지원
스레드 풀 관리
메타데이터 처리

3️⃣ 클라이언트 구현

Python에서 gRPC 클라이언트를 구현하고 서버와 통신하는 방법이다.

import grpc
import logging
import time
from datetime import datetime
import user_pb2
import user_pb2_grpc

# 로깅 설정
logging.basicConfig(level=logging.INFO)

class UserClient:
    def __init__(self, host='localhost', port=50051):
        # 채널 설정 및 스텁 생성
        self.channel = grpc.insecure_channel(f'{host}:{port}')
        self.stub = user_pb2_grpc.UserServiceStub(self.channel)
        logging.info(f"클라이언트가 {host}:{port}에 연결되었습니다")
    
    def __del__(self):
        # 채널 종료
        self.channel.close()
    
    def get_user(self, user_id):
        """사용자 조회"""
        logging.info(f"사용자 조회: {user_id}")
        request = user_pb2.UserRequest(user_id=user_id)
        
        try:
            # 타임아웃 설정 (5초)
            response = self.stub.GetUser(request, timeout=5)
            
            # 응답 처리
            return {
                'user_id': response.user_id,
                'name': response.name,
                'email': response.email,
                'phone': response.phone,
                'age': response.age,
                'created_at': response.created_at,
                'updated_at': response.updated_at
            }
        except grpc.RpcError as e:
            # gRPC 오류 처리
            status_code = e.code()
            details = e.details()
            
            if status_code == grpc.StatusCode.NOT_FOUND:
                logging.warning(f"사용자를 찾을 수 없습니다: {details}")
                return None
            elif status_code == grpc.StatusCode.DEADLINE_EXCEEDED:
                logging.error("요청 시간 초과")
                return None
            else:
                logging.error(f"RPC 오류: {status_code} - {details}")
                return None
        except Exception as e:
            logging.error(f"일반 오류: {str(e)}")
            return None
    
    def create_user(self, name, email, phone=None, age=None):
        """사용자 생성"""
        logging.info(f"사용자 생성: {name}, {email}")
        request = user_pb2.CreateUserRequest(
            name=name,
            email=email,
            phone=phone or "",
            age=age or 0
        )
        
        try:
            # 메타데이터 설정 (예: API 키)
            metadata = [
                ('api-key', 'your-api-key'),
                ('client-id', 'python-client')
            ]
            
            # 타임아웃과 메타데이터로 요청
            response = self.stub.CreateUser(
                request, 
                timeout=5,
                metadata=metadata
            )
            
            # 응답 처리
            return {
                'user_id': response.user_id,
                'name': response.name,
                'email': response.email,
                'phone': response.phone,
                'age': response.age,
                'created_at': response.created_at,
                'updated_at': response.updated_at
            }
        except grpc.RpcError as e:
            status_code = e.code()
            details = e.details()
            
            if status_code == grpc.StatusCode.ALREADY_EXISTS:
                logging.warning(f"이미 존재하는 사용자: {details}")
            elif status_code == grpc.StatusCode.INVALID_ARGUMENT:
                logging.warning(f"잘못된 인자: {details}")
            else:
                logging.error(f"RPC 오류: {status_code} - {details}")
            
            return None
        except Exception as e:
            logging.error(f"일반 오류: {str(e)}")
            return None
    
    def list_users(self, page_size=10, page_number=1, sort_by='name'):
        """사용자 목록 조회"""
        logging.info(f"사용자 목록 조회: 페이지 {page_number}, 크기 {page_size}")
        request = user_pb2.ListUsersRequest(
            page_size=page_size,
            page_number=page_number,
            sort_by=sort_by
        )
        
        try:
            # gRPC 요청
            response = self.stub.ListUsers(request, timeout=10)
            
            # 응답 처리
            users = []
            for user in response.users:
                users.append({
                    'user_id': user.user_id,
                    'name': user.name,
                    'email': user.email,
                    'phone': user.phone,
                    'age': user.age,
                    'created_at': user.created_at,
                    'updated_at': user.updated_at
                })
            
            # 페이지네이션 메타데이터 포함
            return {
                'users': users,
                'total_count': response.total_count,
                'total_pages': response.total_pages,
                'current_page': page_number,
                'page_size': page_size
            }
        except grpc.RpcError as e:
            logging.error(f"RPC 오류: {e.code()} - {e.details()}")
            return None
        except Exception as e:
            logging.error(f"일반 오류: {str(e)}")
            return None

# 클라이언트 사용 예제
def main():
    # 클라이언트 인스턴스 생성
    client = UserClient()
    
    # 새 사용자 생성
    new_user = client.create_user(
        name="홍길동",
        email="[email protected]",
        phone="010-1234-5678",
        age=30
    )
    
    if new_user:
        # 생성된 사용자 ID로 조회
        user_id = new_user['user_id']
        user = client.get_user(user_id)
        
        if user:
            print(f"조회된 사용자: {user['name']}, {user['email']}")
        
        # 사용자 목록 조회
        users_page = client.list_users(page_size=5, page_number=1)
        
        if users_page:
            print(f"총 사용자 수: {users_page['total_count']}")
            for user in users_page['users']:
                print(f"- {user['name']} ({user['email']})")

if __name__ == "__main__":
    main()

✅ 특징:

채널 설정
스텁 생성
에러 처리
메타데이터 전송
타임아웃 관리
응답 파싱
리소스 정리

4️⃣ 스트리밍

service UserService {
    rpc GetUserStream (UserRequest) returns (stream UserResponse) {}
    rpc CreateUsers (stream CreateUserRequest) returns (UsersResponse) {}
    rpc ChatStream (stream ChatMessage) returns (stream ChatMessage) {}
}

class UserServicer(user_pb2_grpc.UserServiceServicer):
    async def GetUserStream(self, request, context):
        users = get_users_batch(request.user_id)
        for user in users:
            yield user_pb2.UserResponse(
                user_id=user.id,
                name=user.name,
                email=user.email
            )

✅ 특징:

서버 스트리밍
클라이언트 스트리밍
양방향 스트리밍

5️⃣ 인터셉터

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        token = metadata.get('authorization')
        
        if not token:
            return self._unauthenticated()
        
        try:
            user = verify_token(token)
            context = set_user_context(user)
            return continuation(handler_call_details, context)
        except Exception:
            return self._unauthenticated()

✅ 특징:

인증/인가
로깅
메타데이터 처리

주요 팁

✅ 모범 사례:

프로토버프 설계 신중히
스트리밍 적절히 활용
에러 처리 표준화
인터셉터 활용
성능 최적화
보안 고려
모니터링 구현
테스트 작성

KR_gRPC - somaz94/python-study GitHub Wiki

Python gRPC 개념 정리

1️⃣ gRPC 기초

gRPC vs REST 비교:

2️⃣ 서버 구현

3️⃣ 클라이언트 구현

4️⃣ 스트리밍

5️⃣ 인터셉터

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_gRPC - somaz94/python-study GitHub Wiki

Python gRPC 개념 정리

1️⃣ gRPC 기초

gRPC vs REST 비교:

2️⃣ 서버 구현

3️⃣ 클라이언트 구현

4️⃣ 스트리밍

5️⃣ 인터셉터

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️