API Gateway

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 예시: FastAPI 라우팅 + 인증 체크, API Gateway 역할 일부 구현
from fastapi import FastAPI, Request, HTTPException, Depends

app = FastAPI()

def verify_auth(request: Request):
    # Gateway에서 인증 토큰 확인
    token = request.headers.get('Authorization')
    if token != "Bearer exampletoken":
        raise HTTPException(status_code=401, detail="Unauthorized")

@app.get("/api/v1/order", dependencies=[Depends(verify_auth)])
def get_order():
    # 주문 서비스와 통신 (실제론 별도 서비스로 라우팅)
    return {"order": "sample_order"}

@app.get("/api/v1/user", dependencies=[Depends(verify_auth)])
def get_user():
    # 유저 서비스와 통신 (실제론 별도 서비스로 라우팅)
    return {"user": "sample_user"}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211

# FastAPI를 이용한 간단한 API Gateway 구현 예시
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.responses import JSONResponse
import httpx
import jwt
import time
from typing import Optional
import redis
import json

app = FastAPI(title="API Gateway", version="1.0.0")

# Redis 클라이언트 (캐싱용)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 서비스 레지스트리 - 실제로는 서비스 디스커버리 사용
SERVICE_REGISTRY = {
    "user": "http://user-service:8001",
    "product": "http://product-service:8002", 
    "order": "http://order-service:8003",
    "payment": "http://payment-service:8004"
}

# JWT 시크릿 키
JWT_SECRET = "your-secret-key"

class RateLimiter:
    """요청 제한 클래스"""
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
    
    def is_allowed(self, client_id: str) -> bool:
        """요청 허용 여부 확인"""
        key = f"rate_limit:{client_id}"
        current_time = int(time.time())
        window_start = current_time - self.window_seconds
        
        # 현재 윈도우의 요청 수 확인
        pipe = redis_client.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)  # 오래된 요청 제거
        pipe.zcard(key)  # 현재 요청 수 확인
        pipe.zadd(key, {str(current_time): current_time})  # 현재 요청 추가
        pipe.expire(key, self.window_seconds)
        
        results = pipe.execute()
        request_count = results[1]
        
        return request_count < self.max_requests

rate_limiter = RateLimiter()

async def verify_token(authorization: Optional[str] = Header(None)):
    """JWT 토큰 검증"""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization header missing")
    
    try:
        # Bearer 토큰 추출
        token = authorization.replace("Bearer ", "")
        payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

async def get_cached_response(cache_key: str):
    """캐시된 응답 조회"""
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            return json.loads(cached_data)
    except Exception:
        pass
    return None

async def set_cached_response(cache_key: str, data: dict, ttl: int = 300):
    """응답 캐싱"""
    try:
        redis_client.setex(cache_key, ttl, json.dumps(data))
    except Exception:
        pass

@app.middleware("http")
async def rate_limiting_middleware(request, call_next):
    """요청 제한 미들웨어"""
    client_ip = request.client.host
    
    if not rate_limiter.is_allowed(client_ip):
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded"}
        )
    
    response = await call_next(request)
    return response

async def proxy_request(service_name: str, path: str, method: str, 
                       headers: dict, body: bytes = None):
    """백엔드 서비스로 요청 프록시"""
    if service_name not in SERVICE_REGISTRY:
        raise HTTPException(status_code=404, detail="Service not found")
    
    service_url = SERVICE_REGISTRY[service_name]
    target_url = f"{service_url}{path}"
    
    # 내부 헤더 제거 (보안상)
    filtered_headers = {k: v for k, v in headers.items() 
                       if not k.lower().startswith('host')}
    
    async with httpx.AsyncClient() as client:
        try:
            if method.upper() == "GET":
                response = await client.get(target_url, headers=filtered_headers)
            elif method.upper() == "POST":
                response = await client.post(target_url, headers=filtered_headers, content=body)
            elif method.upper() == "PUT":
                response = await client.put(target_url, headers=filtered_headers, content=body)
            elif method.upper() == "DELETE":
                response = await client.delete(target_url, headers=filtered_headers)
            else:
                raise HTTPException(status_code=405, detail="Method not allowed")
            
            return response
        except httpx.RequestError:
            raise HTTPException(status_code=503, detail="Service unavailable")

@app.api_route("/api/{service_name}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def gateway_proxy(service_name: str, path: str, request,
                       token_payload: dict = Depends(verify_token)):
    """API Gateway 메인 프록시 엔드포인트"""
    
    # 캐싱 키 생성 (GET 요청만)
    cache_key = None
    if request.method == "GET":
        cache_key = f"cache:{service_name}:{path}:{request.url.query}"
        cached_response = await get_cached_response(cache_key)
        if cached_response:
            return cached_response
    
    # 요청 본문 읽기
    body = await request.body() if request.method in ["POST", "PUT"] else None
    
    # 백엔드 서비스로 요청 전달
    response = await proxy_request(
        service_name=service_name,
        path=f"/{path}",
        method=request.method,
        headers=dict(request.headers),
        body=body
    )
    
    # 응답 처리
    response_data = {
        "status_code": response.status_code,
        "content": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text,
        "headers": dict(response.headers)
    }
    
    # 성공적인 GET 요청 결과 캐싱
    if request.method == "GET" and response.status_code == 200 and cache_key:
        await set_cached_response(cache_key, response_data)
    
    return JSONResponse(
        status_code=response.status_code,
        content=response_data["content"],
        headers={k: v for k, v in response_data["headers"].items() 
                if k.lower() not in ['content-length', 'transfer-encoding']}
    )

@app.get("/health")
async def health_check():
    """헬스 체크 엔드포인트"""
    return {"status": "healthy", "timestamp": time.time()}

@app.get("/metrics")
async def get_metrics():
    """메트릭 수집 엔드포인트"""
    # 실제로는 Prometheus 메트릭 등을 반환
    return {
        "total_requests": "1000",
        "success_rate": "99.5%",
        "avg_response_time": "150ms"
    }

# API Gateway 통합 응답 예시
@app.get("/api/user/{user_id}/profile")
async def get_user_profile(user_id: str, token_payload: dict = Depends(verify_token)):
    """사용자 프로필 통합 조회 - 여러 서비스 어그리게이션 예시"""
    
    async with httpx.AsyncClient() as client:
        # 동시에 여러 서비스 호출
        user_task = client.get(f"{SERVICE_REGISTRY['user']}/users/{user_id}")
        order_task = client.get(f"{SERVICE_REGISTRY['order']}/users/{user_id}/orders/recent")
        
        # 응답 대기
        user_response, order_response = await asyncio.gather(user_task, order_task)
        
        # 응답 통합
        profile_data = {
            "user": user_response.json() if user_response.status_code == 200 else None,
            "recent_orders": order_response.json() if order_response.status_code == 200 else [],
            "profile_complete": user_response.status_code == 200
        }
        
        return profile_data

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# Lambda 함수: 인증 처리
import json

def lambda_handler(event, context):
    token = event['headers'].get('Authorization', '')
    
    # 인증 토큰 검증 (예: JWT 형식)
    if token != "Bearer valid-token":
        return {
            'statusCode': 401,
            'body': json.dumps('Unauthorized')
        }

    # 요청을 백엔드 서비스로 전달할 수 있도록 승인
    return {
        'principalId': 'user123',
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Allow',
                'Resource': event['methodArn']
            }]
        }
    }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256

# api_gateway.py
from flask import Flask, request, jsonify, g
import requests
import jwt
import redis
import time
import logging
from functools import wraps
from datetime import datetime, timedelta

app = Flask(__name__)

# Redis 캐시 설정
cache = redis.Redis(host='localhost', port=6379, db=0)

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

# 서비스 레지스트리 - 실제로는 외부 서비스 디스커버리 사용
SERVICE_REGISTRY = {
    'user-service': 'http://user-service:8001',
    'order-service': 'http://order-service:8002',
    'payment-service': 'http://payment-service:8003',
    'inventory-service': 'http://inventory-service:8004'
}

# JWT 시크릿 키 (실제로는 환경변수에서 로드)
JWT_SECRET = 'your-secret-key'

class APIGateway:
    """API Gateway 메인 클래스"""
    
    def __init__(self):
        self.rate_limit_store = {}  # 속도 제한을 위한 저장소
        
    def authenticate_request(self, f):
        """JWT 토큰 기반 인증 데코레이터"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # Authorization 헤더에서 토큰 추출
            token = request.headers.get('Authorization')
            if not token:
                return jsonify({'error': 'Missing authorization token'}), 401
            
            try:
                # Bearer 접두사 제거
                if token.startswith('Bearer '):
                    token = token[7:]
                
                # JWT 토큰 검증
                payload = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
                g.user_id = payload.get('user_id')
                g.user_role = payload.get('role')
                
                logger.info(f"Authenticated user: {g.user_id}")
                
            except jwt.ExpiredSignatureError:
                return jsonify({'error': 'Token has expired'}), 401
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Invalid token'}), 401
            
            return f(*args, **kwargs)
        return decorated_function
    
    def rate_limit(self, max_requests=60, window=60):
        """속도 제한 데코레이터 (분당 요청 수 제한)"""
        def decorator(f):
            @wraps(f)
            def decorated_function(*args, **kwargs):
                # 클라이언트 IP 또는 사용자 ID 기반으로 제한
                client_id = g.get('user_id') or request.remote_addr
                current_time = int(time.time())
                window_start = current_time - (current_time % window)
                
                # Redis에서 현재 윈도우의 요청 수 확인
                key = f"rate_limit:{client_id}:{window_start}"
                current_requests = cache.get(key)
                
                if current_requests and int(current_requests) >= max_requests:
                    return jsonify({
                        'error': 'Rate limit exceeded',
                        'retry_after': window - (current_time % window)
                    }), 429
                
                # 요청 수 증가
                cache.incr(key)
                cache.expire(key, window)
                
                return f(*args, **kwargs)
            return decorated_function
        return decorator
    
    def cache_response(self, ttl=300):
        """응답 캐싱 데코레이터"""
        def decorator(f):
            @wraps(f)
            def decorated_function(*args, **kwargs):
                # 캐시 키 생성 (URL + 쿼리 파라미터)
                cache_key = f"cache:{request.url}"
                
                # 캐시에서 응답 확인
                cached_response = cache.get(cache_key)
                if cached_response:
                    logger.info(f"Cache hit for: {cache_key}")
                    return jsonify(cached_response.decode())
                
                # 실제 함수 실행
                response = f(*args, **kwargs)
                
                # 성공 응답만 캐싱
                if response[1] == 200:  # status code
                    cache.setex(cache_key, ttl, response[0].data)
                    logger.info(f"Cached response for: {cache_key}")
                
                return response
            return decorated_function
        return decorator
    
    def route_request(self, service_name, path, method='GET'):
        """백엔드 서비스로 요청 라우팅"""
        # 서비스 URL 조회
        service_url = SERVICE_REGISTRY.get(service_name)
        if not service_url:
            return jsonify({'error': f'Service {service_name} not found'}), 404
        
        # 전체 URL 구성
        target_url = f"{service_url}{path}"
        
        try:
            # 요청 헤더 복사 (Authorization 제외)
            headers = {k: v for k, v in request.headers if k != 'Authorization'}
            headers['X-User-ID'] = str(g.get('user_id', ''))
            headers['X-User-Role'] = g.get('user_role', '')
            
            # 백엔드 서비스로 요청 전달
            if method == 'GET':
                response = requests.get(
                    target_url, 
                    params=request.args,
                    headers=headers,
                    timeout=30
                )
            elif method == 'POST':
                response = requests.post(
                    target_url,
                    json=request.get_json(),
                    headers=headers,
                    timeout=30
                )
            elif method == 'PUT':
                response = requests.put(
                    target_url,
                    json=request.get_json(),
                    headers=headers,
                    timeout=30
                )
            elif method == 'DELETE':
                response = requests.delete(
                    target_url,
                    headers=headers,
                    timeout=30
                )
            
            # 응답 로깅
            logger.info(f"Routed {method} {target_url} -> {response.status_code}")
            
            # 응답 반환
            return response.json(), response.status_code
            
        except requests.Timeout:
            logger.error(f"Timeout routing to {service_name}")
            return jsonify({'error': 'Service timeout'}), 504
        except requests.ConnectionError:
            logger.error(f"Connection error routing to {service_name}")
            return jsonify({'error': 'Service unavailable'}), 503
        except Exception as e:
            logger.error(f"Error routing to {service_name}: {str(e)}")
            return jsonify({'error': 'Internal server error'}), 500

# API Gateway 인스턴스 생성
gateway = APIGateway()

# 라우트 정의
@app.route('/api/users/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
@gateway.authenticate_request
@gateway.rate_limit(max_requests=100, window=60)
def user_service_proxy(path):
    """사용자 서비스 프록시"""
    return gateway.route_request('user-service', f"/{path}", request.method)

@app.route('/api/orders/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
@gateway.authenticate_request
@gateway.rate_limit(max_requests=50, window=60)
def order_service_proxy(path):
    """주문 서비스 프록시"""
    return gateway.route_request('order-service', f"/{path}", request.method)

@app.route('/api/products/<path:path>', methods=['GET'])
@gateway.cache_response(ttl=600)  # 10분 캐싱
@gateway.rate_limit(max_requests=200, window=60)
def product_service_proxy(path):
    """상품 서비스 프록시 (읽기 전용, 캐싱 적용)"""
    return gateway.route_request('inventory-service', f"/products/{path}", 'GET')

@app.route('/api/payments/<path:path>', methods=['POST'])
@gateway.authenticate_request
@gateway.rate_limit(max_requests=20, window=60)  # 결제는 더 엄격한 제한
def payment_service_proxy(path):
    """결제 서비스 프록시"""
    return gateway.route_request('payment-service', f"/{path}", request.method)

# 헬스체크 엔드포인트
@app.route('/health')
def health_check():
    """API Gateway 헬스체크"""
    return jsonify({
        'status': 'healthy',
        'timestamp': datetime.utcnow().isoformat(),
        'services': list(SERVICE_REGISTRY.keys())
    })

# API 문서 엔드포인트
@app.route('/api/docs')
def api_documentation():
    """API 문서 제공"""
    docs = {
        'title': 'E-commerce API Gateway',
        'version': '1.0.0',
        'endpoints': {
            '/api/users/*': 'User management operations',
            '/api/orders/*': 'Order management operations',
            '/api/products/*': 'Product catalog (cached)',
            '/api/payments/*': 'Payment processing'
        },
        'authentication': 'JWT Bearer token required',
        'rate_limits': {
            'users': '100 requests/minute',
            'orders': '50 requests/minute',
            'products': '200 requests/minute',
            'payments': '20 requests/minute'
        }
    }
    return jsonify(docs)

# 에러 핸들러
@app.errorhandler(404)
def not_found_error(error):
    return jsonify({'error': 'Endpoint not found'}), 404

@app.errorhandler(500)
def internal_error(error):
    return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=False)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# docker-compose.yml
version: '3.8'
services:
  api-gateway:
    build: .
    ports:
      - "8080:8080"
    environment:
      - JWT_SECRET=your-secret-key
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
      - user-service
      - order-service
      - payment-service
      - inventory-service
    
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    
  user-service:
    image: user-service:latest
    ports:
      - "8001:8001"
    
  order-service:
    image: order-service:latest
    ports:
      - "8002:8002"
    
  payment-service:
    image: payment-service:latest
    ports:
      - "8003:8003"
    
  inventory-service:
    image: inventory-service:latest
    ports:
      - "8004:8004"