Python REST API 개념 정리

1️⃣ REST API 기초

REST(Representational State Transfer)는 웹 서비스를 위한 아키텍처 스타일로, HTTP 프로토콜의 특성을 최대한 활용하여 리소스 중심의 인터페이스를 제공한다.

from flask import Flask, request, jsonify
from flask_restful import Resource, Api
import logging
from typing import Dict, List, Any, Optional

app = Flask(__name__)
api = Api(app)

# 기본 REST 엔드포인트
class UserResource(Resource):
    def get(self, user_id: int) -> Dict[str, Any]:
        """
        사용자 정보 조회
        
        Args:
            user_id: 사용자 ID
            
        Returns:
            dict: 사용자 정보
        """
        # GET 요청 처리 (조회)
        return {'user_id': user_id, 'name': 'Test User', 'email': '[email protected]'}
    
    def post(self) -> tuple:
        """
        새 사용자 생성
        
        Returns:
            tuple: (응답 데이터, HTTP 상태 코드)
        """
        # POST 요청 처리 (생성)
        data = request.get_json()
        # 데이터베이스 저장 로직 구현
        return {'status': 'created', 'user': data}, 201
    
    def put(self, user_id: int) -> Dict[str, Any]:
        """
        사용자 정보 업데이트
        
        Args:
            user_id: 사용자 ID
            
        Returns:
            dict: 업데이트 결과
        """
        # PUT 요청 처리 (전체 업데이트)
        data = request.get_json()
        # 데이터베이스 업데이트 로직 구현
        return {'status': 'updated', 'user_id': user_id}
    
    def patch(self, user_id: int) -> Dict[str, Any]:
        """
        사용자 정보 부분 업데이트
        
        Args:
            user_id: 사용자 ID
            
        Returns:
            dict: 업데이트 결과
        """
        # PATCH 요청 처리 (부분 업데이트)
        data = request.get_json()
        # 데이터베이스 부분 업데이트 로직 구현
        return {'status': 'partially updated', 'user_id': user_id}
    
    def delete(self, user_id: int) -> tuple:
        """
        사용자 삭제
        
        Args:
            user_id: 사용자 ID
            
        Returns:
            tuple: (응답 데이터, HTTP 상태 코드)
        """
        # DELETE 요청 처리 (삭제)
        # 데이터베이스 삭제 로직 구현
        return '', 204

# 리소스 라우팅 설정
api.add_resource(UserResource, '/users', '/users/<int:user_id>')

✅ 특징:

HTTP 메서드 구현 (GET, POST, PUT, PATCH, DELETE)
리소스 기반 설계로 직관적인 API 구조화
REST 원칙에 맞는 상태 코드 활용
URI를 통한 리소스 식별
타입 힌팅을 통한 코드 가독성 향상
메서드 문서화로 API 이해도 증가
명확한 요청/응답 구조
상태를 저장하지 않는 무상태(Stateless) 통신

2️⃣ REST API 클라이언트

API 서버와 통신하기 위한 클라이언트는 requests 라이브러리를 기반으로 구축할 수 있으며, 재사용 가능한 추상화 계층을 제공한다.

import requests
import logging
import json
import time
from typing import Dict, List, Any, Optional, Union
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class RESTClient:
    """REST API와 통신하기 위한 클라이언트 클래스"""
    
    def __init__(
        self,
        base_url: str,
        api_key: Optional[str] = None,
        timeout: int = 30,
        max_retries: int = 3,
        retry_backoff_factor: float = 0.5,
        retry_on_status: List[int] = None
    ):
        """
        REST 클라이언트 초기화
        
        Args:
            base_url: API 기본 URL
            api_key: API 키 (있는 경우)
            timeout: 요청 타임아웃 (초)
            max_retries: 최대 재시도 횟수
            retry_backoff_factor: 재시도 간격 계수 
            retry_on_status: 재시도할 HTTP 상태 코드 목록
        """
        self.logger = logging.getLogger(__name__)
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        
        # 세션 초기화
        self.session = requests.Session()
        
        # 인증 헤더 설정
        if api_key:
            self.session.headers.update({
                'Authorization': f'Bearer {api_key}',
                'Content-Type': 'application/json'
            })
        else:
            self.session.headers.update({
                'Content-Type': 'application/json'
            })
        
        # 재시도 설정
        if retry_on_status is None:
            retry_on_status = [429, 500, 502, 503, 504]
            
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=retry_backoff_factor,
            status_forcelist=retry_on_status,
            allowed_methods=["GET", "POST", "PUT", "DELETE", "PATCH"]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
    def __del__(self):
        """소멸자: 세션 정리"""
        if hasattr(self, 'session'):
            self.session.close()
    
    def _build_url(self, endpoint: str) -> str:
        """
        전체 URL 구성
        
        Args:
            endpoint: API 엔드포인트
            
        Returns:
            str: 완전한 URL
        """
        return f"{self.base_url}/{endpoint.lstrip('/')}"
    
    def _handle_response(self, response: requests.Response) -> Any:
        """
        API 응답 처리
        
        Args:
            response: 요청 응답 객체
            
        Returns:
            Any: 응답 데이터
            
        Raises:
            requests.HTTPError: HTTP 오류 발생 시
        """
        try:
            response.raise_for_status()
            
            if not response.content:
                return None
                
            return response.json()
            
        except json.JSONDecodeError:
            self.logger.warning("응답이 JSON 형식이 아닙니다")
            return response.text
            
        except requests.HTTPError as e:
            self.logger.error(f"HTTP 오류: {e}")
            
            # 오류 응답 내용 추출 시도
            error_detail = None
            try:
                error_detail = response.json()
            except:
                error_detail = response.text
                
            self.logger.error(f"오류 세부 정보: {error_detail}")
            raise
    
    def get(
        self, 
        endpoint: str, 
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Any:
        """
        GET 요청 수행
        
        Args:
            endpoint: API 엔드포인트
            params: 쿼리 파라미터
            headers: 추가 헤더
            
        Returns:
            Any: 응답 데이터
        """
        url = self._build_url(endpoint)
        self.logger.debug(f"GET 요청: {url}")
        
        start_time = time.time()
        response = self.session.get(
            url, 
            params=params, 
            headers=headers,
            timeout=self.timeout
        )
        elapsed = time.time() - start_time
        
        self.logger.debug(f"응답 시간: {elapsed:.2f}초")
        return self._handle_response(response)
    
    def post(
        self, 
        endpoint: str, 
        data: Any,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Any:
        """
        POST 요청 수행
        
        Args:
            endpoint: API 엔드포인트
            data: 요청 데이터
            params: 쿼리 파라미터
            headers: 추가 헤더
            
        Returns:
            Any: 응답 데이터
        """
        url = self._build_url(endpoint)
        self.logger.debug(f"POST 요청: {url}")
        
        start_time = time.time()
        response = self.session.post(
            url, 
            json=data, 
            params=params, 
            headers=headers,
            timeout=self.timeout
        )
        elapsed = time.time() - start_time
        
        self.logger.debug(f"응답 시간: {elapsed:.2f}초")
        return self._handle_response(response)
    
    def put(
        self, 
        endpoint: str, 
        data: Any,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Any:
        """
        PUT 요청 수행
        
        Args:
            endpoint: API 엔드포인트
            data: 요청 데이터
            params: 쿼리 파라미터
            headers: 추가 헤더
            
        Returns:
            Any: 응답 데이터
        """
        url = self._build_url(endpoint)
        self.logger.debug(f"PUT 요청: {url}")
        
        start_time = time.time()
        response = self.session.put(
            url, 
            json=data, 
            params=params, 
            headers=headers,
            timeout=self.timeout
        )
        elapsed = time.time() - start_time
        
        self.logger.debug(f"응답 시간: {elapsed:.2f}초")
        return self._handle_response(response)
    
    def patch(
        self, 
        endpoint: str, 
        data: Any,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Any:
        """
        PATCH 요청 수행
        
        Args:
            endpoint: API 엔드포인트
            data: 요청 데이터
            params: 쿼리 파라미터
            headers: 추가 헤더
            
        Returns:
            Any: 응답 데이터
        """
        url = self._build_url(endpoint)
        self.logger.debug(f"PATCH 요청: {url}")
        
        start_time = time.time()
        response = self.session.patch(
            url, 
            json=data, 
            params=params, 
            headers=headers,
            timeout=self.timeout
        )
        elapsed = time.time() - start_time
        
        self.logger.debug(f"응답 시간: {elapsed:.2f}초")
        return self._handle_response(response)
    
    def delete(
        self, 
        endpoint: str,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Any:
        """
        DELETE 요청 수행
        
        Args:
            endpoint: API 엔드포인트
            params: 쿼리 파라미터
            headers: 추가 헤더
            
        Returns:
            Any: 응답 데이터
        """
        url = self._build_url(endpoint)
        self.logger.debug(f"DELETE 요청: {url}")
        
        start_time = time.time()
        response = self.session.delete(
            url, 
            params=params, 
            headers=headers,
            timeout=self.timeout
        )
        elapsed = time.time() - start_time
        
        self.logger.debug(f"응답 시간: {elapsed:.2f}초")
        return self._handle_response(response)

✅ 특징:

세션 관리로 연결 재사용 최적화
인증 헤더 자동 설정
재시도 메커니즘으로 일시적 오류 처리
타임아웃 설정으로 응답 지연 제한
HTTP 메서드 완전 지원 (GET, POST, PUT, PATCH, DELETE)
에러 처리 및 로깅 통합
응답 시간 측정 및 디버깅
JSON 변환 자동화
타입 힌팅을 통한 코드 가독성 향상
리소스 정리를 위한 소멸자 구현

3️⃣ API 인증과 보안

REST API에서 인증과 보안은 중요한 고려사항으로, 다양한 방식의 인증 메커니즘과 보안 기법을 통해 API 리소스를 보호할 수 있다.

from flask import Flask, request, jsonify
from flask_restful import Resource, Api
from functools import wraps
import jwt
import datetime
import uuid
import logging
import hashlib
from typing import Dict, Any, Callable, Optional

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'  # 실제 환경에서는 환경 변수 등으로 관리
api = Api(app)

# 로깅 설정
logger = logging.getLogger(__name__)

# JWT 토큰 생성
def generate_token(user_id: int, expiry_minutes: int = 60) -> str:
    """
    JWT 토큰 생성 함수
    
    Args:
        user_id: 사용자 ID
        expiry_minutes: 토큰 만료 시간(분)
        
    Returns:
        str: 생성된 JWT 토큰
    """
    payload = {
        'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=expiry_minutes),
        'iat': datetime.datetime.utcnow(),
        'sub': user_id,
        'jti': str(uuid.uuid4())
    }
    
    token = jwt.encode(
        payload,
        app.config['SECRET_KEY'],
        algorithm='HS256'
    )
    
    return token

# 인증 데코레이터
def require_auth(f: Callable) -> Callable:
    """
    인증이 필요한 API 엔드포인트에 사용하는 데코레이터
    
    Args:
        f: 원본 함수
        
    Returns:
        Callable: 래핑된 함수
    """
    @wraps(f)
    def decorated(*args: Any, **kwargs: Any) -> Any:
        # 인증 헤더 확인
        auth_header = request.headers.get('Authorization')
        if not auth_header:
            logger.warning("인증 토큰 없음")
            return {'message': '인증 토큰이 필요합니다', 'error': 'unauthorized'}, 401
        
        # Bearer 토큰 형식 확인
        parts = auth_header.split()
        if parts[0].lower() != 'bearer' or len(parts) != 2:
            logger.warning("잘못된 인증 헤더 형식")
            return {'message': '잘못된 인증 헤더 형식', 'error': 'invalid_header'}, 401
        
        token = parts[1]
        
        try:
            # 토큰 검증
            payload = jwt.decode(
                token, 
                app.config['SECRET_KEY'],
                algorithms=['HS256']
            )
            
            # 사용자 정보 설정
            request.user_id = payload['sub']
            
            return f(*args, **kwargs)
            
        except jwt.ExpiredSignatureError:
            logger.warning(f"만료된 토큰: {token[:10]}...")
            return {'message': '토큰이 만료되었습니다', 'error': 'token_expired'}, 401
            
        except jwt.InvalidTokenError:
            logger.warning(f"유효하지 않은 토큰: {token[:10]}...")
            return {'message': '유효하지 않은 토큰입니다', 'error': 'invalid_token'}, 401
            
        except Exception as e:
            logger.error(f"인증 처리 중 오류: {str(e)}")
            return {'message': '서버 인증 오류', 'error': 'auth_error'}, 500
    
    return decorated

# API 속도 제한 데코레이터
def rate_limit(limit: int = 100, period: int = 3600) -> Callable:
    """
    API 요청 속도 제한 데코레이터
    
    Args:
        limit: 허용 요청 수
        period: 시간 기간(초)
        
    Returns:
        Callable: 데코레이터 함수
    """
    def decorator(f: Callable) -> Callable:
        # 실제 구현에서는 Redis 등의 분산 캐시 사용 권장
        request_counts = {}
        
        @wraps(f)
        def decorated(*args: Any, **kwargs: Any) -> Any:
            # 클라이언트 식별
            client_ip = request.remote_addr
            client_id = request.headers.get('X-API-Key', client_ip)
            current_time = int(datetime.datetime.utcnow().timestamp())
            
            # 요청 수 관리
            if client_id not in request_counts:
                request_counts[client_id] = {'count': 1, 'reset_time': current_time + period}
            else:
                if current_time > request_counts[client_id]['reset_time']:
                    # 기간 초기화
                    request_counts[client_id] = {'count': 1, 'reset_time': current_time + period}
                else:
                    # 요청 수 증가
                    request_counts[client_id]['count'] += 1
            
            # 한도 확인
            if request_counts[client_id]['count'] > limit:
                reset_time = request_counts[client_id]['reset_time']
                reset_in = reset_time - current_time
                
                logger.warning(f"속도 제한 초과: {client_id}")
                
                # 속도 제한 응답 헤더
                response = jsonify({
                    'message': '요청 한도 초과',
                    'error': 'rate_limit_exceeded'
                })
                response.status_code = 429
                response.headers['X-RateLimit-Limit'] = str(limit)
                response.headers['X-RateLimit-Remaining'] = '0'
                response.headers['X-RateLimit-Reset'] = str(reset_time)
                response.headers['Retry-After'] = str(reset_in)
                
                return response
            
            # 요청 처리
            return f(*args, **kwargs)
        
        return decorated
    return decorator

# 보호된 리소스 예제
class SecureResource(Resource):
    @require_auth
    @rate_limit(limit=5, period=60)  # 1분당 5개 요청으로 제한
    def get(self) -> Dict[str, Any]:
        """보호된 리소스 조회"""
        return {
            'message': '보호된 리소스',
            'user_id': request.user_id,
            'timestamp': datetime.datetime.utcnow().isoformat()
        }

# 인증 엔드포인트
class AuthResource(Resource):
    def post(self) -> Dict[str, Any]:
        """사용자 인증 및 토큰 발급"""
        data = request.get_json()
        
        # 간단한 인증 예제 (실제 환경에서는 DB 인증 사용)
        username = data.get('username')
        password = data.get('password')
        
        # 비밀번호 해싱 처리 예제
        password_hash = hashlib.sha256(password.encode()).hexdigest()
        
        # 사용자 검증 (예시)
        if username == 'admin' and password_hash == hashlib.sha256('password'.encode()).hexdigest():
            # 토큰 생성
            token = generate_token(user_id=1)
            
            return {
                'access_token': token,
                'token_type': 'Bearer',
                'expires_in': 3600
            }
        else:
            return {'message': '인증 실패', 'error': 'invalid_credentials'}, 401

# 엔드포인트 등록
api.add_resource(SecureResource, '/secure')
api.add_resource(AuthResource, '/auth')

✅ 특징:

JWT 기반 인증 시스템 구현
토큰 생성 및 검증 메커니즘
데코레이터 패턴을 활용한 인증 로직 분리
속도 제한(Rate Limiting) 구현
토큰 만료 및 갱신 관리
로깅을 통한 보안 이벤트 추적
상세한 오류 메시지 및 코드
HTTP 헤더를 활용한 인증 정보 전달
보안 헤더 설정
비밀번호 해싱 처리

4️⃣ API 버전 관리

API 버전 관리는 API 변경사항을 도입하면서 기존 클라이언트와의 호환성을 유지하기 위한 필수적인 전략이다. 여러 버전 관리 방식을 적절히 조합하여 사용할 수 있다.

from flask import Flask, request, jsonify, Blueprint
from flask_restful import Resource, Api, reqparse
import logging
from typing import Dict, Any, Union, Tuple

app = Flask(__name__)
api = Api(app)

logger = logging.getLogger(__name__)

# 1. URL 경로 기반 버전 관리
class APIv1(Resource):
    """API 버전 1 (URL 기반)"""
    
    def get(self) -> Dict[str, Any]:
        """V1 데이터 조회"""
        return {
            'version': '1.0',
            'data': 'This is version 1 data',
            'features': ['basic_feature']
        }
        
class APIv2(Resource):
    """API 버전 2 (URL 기반)"""
    
    def get(self) -> Dict[str, Any]:
        """V2 데이터 조회 (향상된 기능)"""
        return {
            'version': '2.0',
            'data': 'This is version 2 data with enhancements',
            'features': ['basic_feature', 'enhanced_feature']
        }

# URL에 버전 정보 포함
api.add_resource(APIv1, '/v1/resource')
api.add_resource(APIv2, '/v2/resource')

# 2. 헤더 기반 버전 관리
class VersionedHeaderResource(Resource):
    """헤더 기반 버전 관리"""
    
    def get(self) -> Union[Dict[str, Any], Tuple[Dict[str, Any], int]]:
        """버전별 응답 반환"""
        # API 버전 헤더 확인
        version = request.headers.get('API-Version', '1.0')
        
        logger.info(f"요청된 API 버전: {version}")
        
        if version == '1.0':
            return {
                'version': '1.0',
                'data': 'This is version 1 data via header',
                'features': ['basic_feature']
            }
        elif version == '2.0':
            return {
                'version': '2.0',
                'data': 'This is version 2 data via header with enhancements',
                'features': ['basic_feature', 'enhanced_feature']
            }
        else:
            # 지원하지 않는 버전
            return {
                'error': 'unsupported_version',
                'message': f'버전 {version}은 지원되지 않습니다. 지원 버전: 1.0, 2.0'
            }, 400

# 헤더 기반 버전 관리용 엔드포인트
api.add_resource(VersionedHeaderResource, '/resource')

# 3. 쿼리 파라미터 기반 버전 관리
class VersionedQueryResource(Resource):
    """쿼리 파라미터 기반 버전 관리"""
    
    def get(self) -> Union[Dict[str, Any], Tuple[Dict[str, Any], int]]:
        """버전별 응답 반환"""
        # 쿼리 파라미터에서 버전 확인
        parser = reqparse.RequestParser()
        parser.add_argument('version', type=str, default='1.0')
        args = parser.parse_args()
        
        version = args['version']
        logger.info(f"쿼리 파라미터 API 버전: {version}")
        
        if version == '1.0':
            return {
                'version': '1.0',
                'data': 'This is version 1 data via query parameter',
                'features': ['basic_feature']
            }
        elif version == '2.0':
            return {
                'version': '2.0',
                'data': 'This is version 2 data via query parameter with enhancements',
                'features': ['basic_feature', 'enhanced_feature']
            }
        else:
            # 지원하지 않는 버전
            return {
                'error': 'unsupported_version',
                'message': f'버전 {version}은 지원되지 않습니다. 지원 버전: 1.0, 2.0'
            }, 400

# 쿼리 파라미터 기반 버전 관리용 엔드포인트
api.add_resource(VersionedQueryResource, '/query-resource')

# 4. 콘텐츠 협상 기반 버전 관리
class ContentNegotiationResource(Resource):
    """콘텐츠 협상(Accept 헤더) 기반 버전 관리"""
    
    def get(self) -> Union[Dict[str, Any], Tuple[Dict[str, Any], int]]:
        """버전별 응답 반환"""
        # Accept 헤더 확인
        accept_header = request.headers.get('Accept', 'application/json')
        logger.info(f"Accept 헤더: {accept_header}")
        
        if accept_header == 'application/vnd.api.v1+json':
            return {
                'version': '1.0',
                'data': 'This is version 1 data via content negotiation',
                'features': ['basic_feature']
            }
        elif accept_header == 'application/vnd.api.v2+json':
            return {
                'version': '2.0',
                'data': 'This is version 2 data via content negotiation with enhancements',
                'features': ['basic_feature', 'enhanced_feature']
            }
        else:
            # 기본 응답은 최신 버전으로
            return {
                'version': '2.0',
                'data': 'This is default version data',
                'features': ['basic_feature', 'enhanced_feature']
            }

# 콘텐츠 협상 기반 버전 관리용 엔드포인트
api.add_resource(ContentNegotiationResource, '/content-resource')

# 5. 블루프린트를 사용한 버전 관리 (모듈화)
api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

# 블루프린트별 API 인스턴스 생성
api_v1_instance = Api(api_v1)
api_v2_instance = Api(api_v2)

# V1 리소스 정의
class UserResourceV1(Resource):
    def get(self, user_id: int) -> Dict[str, Any]:
        """V1 사용자 정보 조회"""
        return {
            'id': user_id,
            'name': 'User Name',
            'version': '1.0'
        }

# V2 리소스 정의 (새 필드 추가)
class UserResourceV2(Resource):
    def get(self, user_id: int) -> Dict[str, Any]:
        """V2 사용자 정보 조회 (추가 필드)"""
        return {
            'id': user_id,
            'name': 'User Name',
            'email': '[email protected]',  # 새 필드
            'profile_url': f'/api/v2/users/{user_id}/profile',  # 새 필드
            'version': '2.0'
        }

# 블루프린트에 리소스 등록
api_v1_instance.add_resource(UserResourceV1, '/users/<int:user_id>')
api_v2_instance.add_resource(UserResourceV2, '/users/<int:user_id>')

# 앱에 블루프린트 등록
app.register_blueprint(api_v1)
app.register_blueprint(api_v2)

# 버전 관리 정책 및 설명 페이지
@app.route('/api/versions')
def api_versions() -> Dict[str, Any]:
    """API 버전 정보 및 설명 제공"""
    return {
        'versions': [
            {
                'version': '1.0',
                'status': 'stable',
                'released': '2022-01-01',
                'sunset_date': '2023-12-31',
                'documentation': '/docs/v1'
            },
            {
                'version': '2.0',
                'status': 'current',
                'released': '2023-01-01',
                'sunset_date': None,
                'documentation': '/docs/v2'
            }
        ],
        'versioning_methods': [
            'URL 경로 (/v1/resource, /v2/resource)',
            'API-Version 헤더',
            '쿼리 파라미터 (?version=1.0)',
            'Accept 헤더 (application/vnd.api.v1+json)'
        ],
        'deprecation_policy': 'API 버전은 출시일로부터 최소 1년간 지원됩니다.'
    }

✅ 특징:

다양한 버전 관리 전략 구현
- URL 경로 기반 버전 관리 (/v1/resource)
- HTTP 헤더 기반 버전 관리 (API-Version)
- 쿼리 파라미터 기반 버전 관리 (?version=1.0)
- 콘텐츠 협상 기반 버전 관리 (Accept 헤더)
블루프린트를 활용한 모듈식 버전 관리
API 버전 정보 및 지원 정책 제공
버전별 기능 차이 명시
버전 지원 종료 일정 관리
타입 힌팅을 통한 코드 가독성 향상
로깅을 통한 버전 요청 추적
지원하지 않는 버전에 대한 오류 처리
기본 버전 설정으로 하위 호환성 유지

5️⃣ API 문서화와 자동화

REST API의 성공적인 사용 및 도입을 위해서는 명확하고 이해하기 쉬운 문서가 필수적이다. 자동화된 문서 생성 도구를 활용하면 코드와 문서의 일관성을 유지할 수 있다.

from flask import Flask, request, jsonify
from flask_restful import Api, Resource
from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from apispec_webframeworks.flask import FlaskPlugin
from marshmallow import Schema, fields
import logging
from typing import Dict, List, Any, Optional

app = Flask(__name__)
api = Api(app)

# APISpec 설정
spec = APISpec(
    title="Sample REST API",
    version="1.0.0",
    openapi_version="3.0.2",
    plugins=[FlaskPlugin(), MarshmallowPlugin()],
    info={
        "description": "샘플 REST API 문서",
        "contact": {"email": "[email protected]"}
    }
)

# 스키마 정의
class UserSchema(Schema):
    """사용자 데이터 스키마"""
    id = fields.Int(required=True, description="사용자 ID")
    name = fields.Str(required=True, description="이름")
    email = fields.Email(required=True, description="이메일 주소")
    created_at = fields.DateTime(description="생성 시간")
    
class ErrorSchema(Schema):
    """오류 응답 스키마"""
    error = fields.Str(required=True, description="오류 코드")
    message = fields.Str(required=True, description="오류 메시지")

# API 리소스 정의
class UserResource(Resource):
    def get(self, user_id):
        """
        사용자 정보 조회
        ---
        parameters:
          - in: path
            name: user_id
            schema:
              type: integer
            required: true
            description: 사용자 ID
        responses:
          200:
            description: 성공
            content:
              application/json:
                schema: UserSchema
          404:
            description: 사용자를 찾을 수 없음
            content:
              application/json:
                schema: ErrorSchema
        """
        return {
            "id": user_id,
            "name": "Example User",
            "email": "[email protected]",
            "created_at": "2023-01-01T00:00:00Z"
        }
    
    def put(self, user_id):
        """
        사용자 정보 업데이트
        ---
        parameters:
          - in: path
            name: user_id
            schema:
              type: integer
            required: true
            description: 사용자 ID
        requestBody:
          content:
            application/json:
              schema: UserSchema
        responses:
          200:
            description: 성공적으로 업데이트됨
            content:
              application/json:
                schema: UserSchema
          400:
            description: 잘못된 요청
            content:
              application/json:
                schema: ErrorSchema
          404:
            description: 사용자를 찾을 수 없음
            content:
              application/json:
                schema: ErrorSchema
        """
        data = request.get_json()
        return {
            "id": user_id,
            "name": data.get("name", "Updated User"),
            "email": data.get("email", "[email protected]"),
            "created_at": "2023-01-01T00:00:00Z"
        }

# 리소스 등록
api.add_resource(UserResource, '/users/<int:user_id>')

# 스키마 등록
spec.components.schema("User", schema=UserSchema)
spec.components.schema("Error", schema=ErrorSchema)

# 경로 등록
with app.test_request_context():
    spec.path(view=UserResource, path="/users/{user_id}", operations={"get": {}, "put": {}})

# Swagger UI 문서 제공 엔드포인트
@app.route('/swagger.json')
def create_swagger_spec():
    """OpenAPI 스펙 생성"""
    return jsonify(spec.to_dict())

# Swagger UI 페이지 제공
@app.route('/docs')
def swagger_ui():
    """Swagger UI HTML 페이지 제공"""
    return f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>API 문서</title>
        <meta charset="utf-8">
        <link rel="stylesheet" type="text/css" href="https://unpkg.com/swagger-ui-dist@3/swagger-ui.css">
    </head>
    <body>
        <div id="swagger-ui"></div>
        <script src="https://unpkg.com/swagger-ui-dist@3/swagger-ui-bundle.js"></script>
        <script>
            const ui = SwaggerUIBundle({{
                url: '/swagger.json',
                dom_id: '#swagger-ui',
                presets: [
                    SwaggerUIBundle.presets.apis,
                    SwaggerUIBundle.SwaggerUIStandalonePreset
                ],
                layout: "BaseLayout",
                deepLinking: true
            }})
        </script>
    </body>
    </html>
    """

# 문서화 도구와 코드 생성 예제
def generate_client_code(spec_path: str, language: str) -> str:
    """
    OpenAPI 스펙에서 클라이언트 코드 생성 (예시)
    
    Args:
        spec_path: OpenAPI 스펙 파일 경로
        language: 생성할 언어 (python, javascript, etc.)
        
    Returns:
        str: 생성된 코드 저장 경로
    """
    # 실제 구현에서는 도구 호출 (예: OpenAPI Generator)
    supported = ["python", "javascript", "typescript", "java", "csharp"]
    
    if language not in supported:
        raise ValueError(f"지원하지 않는 언어: {language}. 지원 언어: {', '.join(supported)}")
    
    # 예시 코드 (실제 구현은 도구 사용)
    print(f"{language} 클라이언트 코드 생성 중...")
    output_path = f"./generated-clients/{language}-client"
    
    # 실제 구현:
    # subprocess.run(["openapi-generator", "generate", "-i", spec_path, 
    #                 "-g", language, "-o", output_path])
    
    return output_path

✅ 특징:

OpenAPI/Swagger 명세를 통한 문서화
apispec을 활용한 동적 문서 생성
Marshmallow 스키마와 통합된 데이터 검증
Swagger UI를 통한 대화형 문서
도큐멘테이션 주석 활용
자동화된 클라이언트 코드 생성
요청/응답 스키마 정의
API 엔드포인트 메타데이터 정의
코드와 문서의 일관성 유지
문서 자동 업데이트 메커니즘

주요 팁

✅ 모범 사례:

리소스 중심 설계: URI는 명사형 리소스를 나타내며, HTTP 메서드는 동작을 나타내도록 설계한다. (예: /users, /articles)
일관된 명명 규칙: URI 경로, 쿼리 파라미터, 응답 필드에 일관된 명명 규칙을 적용한다. (camelCase 또는 snake_case 중 하나를 선택하여 통일)
적절한 HTTP 상태 코드: 각 응답에 적합한 HTTP 상태 코드를 사용한다. (200 OK, 201 Created, 400 Bad Request, 401 Unauthorized, 404 Not Found 등)
요청 검증: 모든 API 입력값을 검증하고 유효하지 않은 경우 명확한 오류 메시지를 반환한다.
응답 형식 표준화: 모든 API 응답에 일관된 형식을 사용하여 클라이언트가 쉽게 처리할 수 있도록 한다.
페이지네이션 구현: 대량의 데이터를 반환하는 API에는 페이지네이션을 적용하고, 링크 헤더나 메타데이터를 통해 탐색 정보를 제공한다.
부분 응답 지원: 클라이언트가 필요한 필드만 요청할 수 있는 메커니즘을 제공하여 네트워크 부하를 줄인다.
API 버전 관리: 호환성을 깨는 변경사항이 있을 때 버전을 명확히 관리하고, 이전 버전에 대한 지원 정책을 문서화한다.
CORS 설정: 웹 애플리케이션이 API를 사용할 수 있도록 적절한 CORS(Cross-Origin Resource Sharing) 헤더를 설정한다.
적절한 인증 메커니즘: API 용도와 보안 요구사항에 맞는 인증 메커니즘을 선택한다. (JWT, OAuth 2.0, API 키 등)
속도 제한 구현: 서비스 남용을 방지하기 위해 클라이언트별 요청 속도 제한을 구현하고, 제한 정보를 헤더에 포함한다.
캐싱 전략: Cache-Control, ETag 등의 헤더를 사용하여 효율적인 캐싱을 구현하고 불필요한 요청을 줄인다.
HATEOAS 고려: 필요한 경우 응답에 관련 리소스 링크를 포함하여 API 탐색성을 향상시킨다. (Hypermedia as the Engine of Application State)
효율적인 배치 작업: 다수의 리소스를 처리해야 하는 경우 배치 작업을 지원하여 네트워크 요청 수를 줄인다.
비동기 작업 처리: 시간이 오래 걸리는 작업은 비동기로 처리하고 상태 확인 엔드포인트를 제공한다.
상세한 로깅: API 요청과 응답을 적절히 로깅하여 디버깅과 모니터링을 용이하게 한다.
문서화: 모든 API 엔드포인트, 파라미터, 응답 형식, 오류 코드를 상세히 문서화한다.
API 헬스체크 엔드포인트: API 상태를 모니터링할 수 있는 헬스체크 엔드포인트를 제공한다.
보안 헤더 설정: X-Content-Type-Options, X-Frame-Options, Content-Security-Policy 등의 보안 헤더를 설정한다.

KR_REST - somaz94/python-study GitHub Wiki

Python REST API 개념 정리

1️⃣ REST API 기초

2️⃣ REST API 클라이언트

3️⃣ API 인증과 보안

4️⃣ API 버전 관리

5️⃣ API 문서화와 자동화

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_REST - somaz94/python-study GitHub Wiki

Python REST API 개념 정리

1️⃣ REST API 기초

2️⃣ REST API 클라이언트

3️⃣ API 인증과 보안

4️⃣ API 버전 관리

5️⃣ API 문서화와 자동화

주요 팁

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

⚠️ GitHub.com Fallback ⚠️