먼저 필요한 패키지를 설치해야 해:

pip install fastapi uvicorn sqlalchemy pydantic

이제 간단한 사용자 서비스를 만들어보자:

# main.py
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
import uvicorn

# 데이터베이스 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 데이터베이스 모델
class UserDB(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)

# Pydantic 모델
class UserCreate(BaseModel):
    username: str
    email: str
    full_name: str

class User(BaseModel):
    id: int
    username: str
    email: str
    full_name: str
    
    class Config:
        orm_mode = True

# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)

# FastAPI 앱 생성
app = FastAPI(title="User Microservice")

# 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# API 엔드포인트
@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = UserDB(username=user.username, email=user.email, full_name=user.full_name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(UserDB).offset(skip).limit(limit).all()
    return users

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

이 코드를 실행하면 http://localhost:8000/docs에서 자동 생성된 API 문서를 확인할 수 있어!

Dockerfile 생성:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
      

requirements.txt 파일:

fastapi==0.110.0
uvicorn==0.27.1
sqlalchemy==2.0.27
pydantic==2.5.3
      

Docker 이미지 빌드 및 실행:

# 이미지 빌드
docker build -t user-service .

# 컨테이너 실행
docker run -d -p 8000:8000 --name user-service user-service
      

# api_gateway.py
from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.responses import JSONResponse
import httpx
import jwt
from typing import Optional

app = FastAPI(title="API Gateway")

# 서비스 URL 설정
SERVICE_URLS = {
    "user": "http://user-service:8000",
    "product": "http://product-service:8001",
    "order": "http://order-service:8002",
    "payment": "http://payment-service:8003"
}

# JWT 시크릿 키
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# 인증 함수
async def verify_token(authorization: Optional[str] = Header(None)):
    if authorization is None:
        raise HTTPException(status_code=401, detail="Authorization header missing")
    
    try:
        token = authorization.split(" ")[1]
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# 사용자 서비스 라우팅
@app.get("/api/users/{user_id}")
async def get_user(user_id: int, token_data: dict = Depends(verify_token)):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{SERVICE_URLS['user']}/users/{user_id}")
        return JSONResponse(content=response.json(), status_code=response.status_code)

@app.post("/api/users/")
async def create_user(token_data: dict = Depends(verify_token)):
    async with httpx.AsyncClient() as client:
        response = await client.post(f"{SERVICE_URLS['user']}/users/")
        return JSONResponse(content=response.json(), status_code=response.status_code)

# 상품 서비스 라우팅
@app.get("/api/products/{product_id}")
async def get_product(product_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{SERVICE_URLS['product']}/products/{product_id}")
        return JSONResponse(content=response.json(), status_code=response.status_code)

# 주문 서비스 라우팅
@app.post("/api/orders/")
async def create_order(token_data: dict = Depends(verify_token)):
    async with httpx.AsyncClient() as client:
        response = await client.post(f"{SERVICE_URLS['order']}/orders/")
        return JSONResponse(content=response.json(), status_code=response.status_code)

# 결제 서비스 라우팅
@app.post("/api/payments/")
async def process_payment(token_data: dict = Depends(verify_token)):
    async with httpx.AsyncClient() as client:
        response = await client.post(f"{SERVICE_URLS['payment']}/payments/")
        return JSONResponse(content=response.json(), status_code=response.status_code)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("api_gateway:app", host="0.0.0.0", port=8080, reload=True)
      

# service_discovery.py
import consul
import socket
import uuid
import time

class ServiceRegistry:
    def __init__(self, service_name, service_port, consul_host="localhost", consul_port=8500):
        self.service_name = service_name
        self.service_port = service_port
        self.consul_client = consul.Consul(host=consul_host, port=consul_port)
        self.service_id = f"{service_name}-{uuid.uuid4()}"
        self.hostname = socket.gethostname()
        self.ip_address = socket.gethostbyname(self.hostname)
    
    def register(self):
        """서비스를 Consul에 등록"""
        self.consul_client.agent.service.register(
            name=self.service_name,
            service_id=self.service_id,
            address=self.ip_address,
            port=self.service_port,
            check=consul.Check.http(f"http://{self.ip_address}:{self.service_port}/health", "10s")
        )
        print(f"Service {self.service_name} registered with ID {self.service_id}")
    
    def deregister(self):
        """서비스를 Consul에서 제거"""
        self.consul_client.agent.service.deregister(self.service_id)
        print(f"Service {self.service_id} deregistered")
    
    @staticmethod
    def discover_service(service_name, consul_host="localhost", consul_port=8500):
        """서비스 인스턴스 찾기"""
        consul_client = consul.Consul(host=consul_host, port=consul_port)
        services = consul_client.catalog.service(service_name)[1]
        if not services:
            return None
        
        # 간단한 로드 밸런싱 (라운드 로빈)
        service = services[int(time.time()) % len(services)]
        return {
            "address": service["ServiceAddress"],
            "port": service["ServicePort"]
        }

# 사용 예시
if __name__ == "__main__":
    # 서비스 등록
    registry = ServiceRegistry("user-service", 8000)
    registry.register()
    
    try:
        # 서비스 실행 중...
        while True:
            time.sleep(10)
            print("Service running...")
    except KeyboardInterrupt:
        # 서비스 종료 시 등록 해제
        registry.deregister()
    
    # 다른 서비스에서 사용자 서비스 찾기
    user_service = ServiceRegistry.discover_service("user-service")
    if user_service:
        print(f"Found user service at {user_service['address']}:{user_service['port']}")
    else:
        print("User service not found")
      

# circuit_breaker.py
import pybreaker
import httpx
import asyncio
from fastapi import FastAPI, HTTPException

app = FastAPI(title="Circuit Breaker Example")

# 서킷 브레이커 설정
# - max_failures: 서킷이 열리기 전 허용되는 최대 실패 횟수
# - reset_timeout: 서킷이 half-open 상태로 전환되기 전 대기 시간(초)
user_service_breaker = pybreaker.CircuitBreaker(
    fail_max=3,
    reset_timeout=30,
    exclude=[HTTPException]
)

# 서킷 브레이커 이벤트 리스너
@user_service_breaker.on_open
def on_open():
    print("Circuit breaker opened - user service is down!")

@user_service_breaker.on_close
def on_close():
    print("Circuit breaker closed - user service is operational again!")

@user_service_breaker.on_half_open
def on_half_open():
    print("Circuit breaker half-open - trying to test if user service is back")

# 서킷 브레이커로 보호된 함수
@user_service_breaker
async def get_user_from_service(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service:8000/users/{user_id}", timeout=2.0)
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code, detail=response.text)
        return response.json()

# 폴백(fallback) 메커니즘
async def get_user_fallback(user_id: int):
    # 캐시에서 데이터 가져오기 또는 기본값 반환
    return {
        "id": user_id,
        "username": "unknown",
        "email": "unknown@example.com",
        "full_name": "Unknown User"
    }

# API 엔드포인트
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        # 서킷 브레이커로 보호된 함수 호출
        return await get_user_from_service(user_id)
    except pybreaker.CircuitBreakerError:
        # 서킷이 열려 있을 때 폴백 메커니즘 사용
        print(f"Circuit is OPEN! Using fallback for user {user_id}")
        return await get_user_fallback(user_id)
    except Exception as e:
        # 다른 예외 처리
        print(f"Error getting user {user_id}: {str(e)}")
        return await get_user_fallback(user_id)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("circuit_breaker:app", host="0.0.0.0", port=8081, reload=True)
      

# order_service.py (동기식 통신 예제)
from fastapi import FastAPI, HTTPException
import httpx
from pydantic import BaseModel

app = FastAPI(title="Order Service")

class OrderCreate(BaseModel):
    user_id: int
    product_id: int
    quantity: int

class Order(BaseModel):
    id: int
    user_id: int
    product_id: int
    quantity: int
    total_price: float

# 동기식으로 다른 서비스 호출
async def get_product_price(product_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://product-service:8001/products/{product_id}")
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code, detail="Product service error")
        product = response.json()
        return product["price"]

async def check_user_exists(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service:8000/users/{user_id}")
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code, detail="User service error")
        return True

@app.post("/orders/", response_model=Order)
async def create_order(order: OrderCreate):
    # 사용자 확인 (동기식 호출)
    await check_user_exists(order.user_id)
    
    # 상품 가격 확인 (동기식 호출)
    price = await get_product_price(order.product_id)
    
    # 주문 생성
    total_price = price * order.quantity
    new_order = {
        "id": 1,  # 실제로는 DB에서 생성된 ID
        "user_id": order.user_id,
        "product_id": order.product_id,
        "quantity": order.quantity,
        "total_price": total_price
    }
    
    # 실제 구현에서는 DB에 저장
    
    return new_order
      

먼저 Protocol Buffers로 서비스를 정의해야 해:

// product.proto
syntax = "proto3";

package product;

service ProductService {
  rpc GetProduct (ProductRequest) returns (ProductResponse) {}
}

message ProductRequest {
  int32 product_id = 1;
}

message ProductResponse {
  int32 id = 1;
  string name = 2;
  float price = 3;
  int32 stock = 4;
}
      

이제 서버 측 구현:

# product_service.py (gRPC 서버)
import grpc
from concurrent import futures
import product_pb2
import product_pb2_grpc

class ProductServicer(product_pb2_grpc.ProductServiceServicer):
    def GetProduct(self, request, context):
        # 실제 구현에서는 데이터베이스에서 제품 정보를 가져옴
        product_id = request.product_id
        
        # 예시 데이터
        products = {
            1: {"id": 1, "name": "Laptop", "price": 1299.99, "stock": 10},
            2: {"id": 2, "name": "Smartphone", "price": 699.99, "stock": 20},
            3: {"id": 3, "name": "Headphones", "price": 149.99, "stock": 30}
        }
        
        if product_id not in products:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"Product with ID {product_id} not found")
            return product_pb2.ProductResponse()
        
        product = products[product_id]
        return product_pb2.ProductResponse(
            id=product["id"],
            name=product["name"],
            price=product["price"],
            stock=product["stock"]
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    product_pb2_grpc.add_ProductServiceServicer_to_server(ProductServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Product gRPC server started on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()
      

클라이언트 측 구현:

# order_service_grpc_client.py
import grpc
import product_pb2
import product_pb2_grpc
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Order Service (gRPC Client)")

class OrderCreate(BaseModel):
    user_id: int
    product_id: int
    quantity: int

class Order(BaseModel):
    id: int
    user_id: int
    product_id: int
    quantity: int
    total_price: float

# gRPC를 사용하여 제품 서비스 호출
def get_product_price(product_id: int):
    with grpc.insecure_channel('product-service:50051') as channel:
        stub = product_pb2_grpc.ProductServiceStub(channel)
        request = product_pb2.ProductRequest(product_id=product_id)
        try:
            response = stub.GetProduct(request)
            return response.price
        except grpc.RpcError as e:
            status_code = e.code()
            if status_code == grpc.StatusCode.NOT_FOUND:
                raise HTTPException(status_code=404, detail=f"Product {product_id} not found")
            else:
                raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")

@app.post("/orders/", response_model=Order)
def create_order(order: OrderCreate):
    # 제품 가격 확인 (gRPC 호출)
    price = get_product_price(order.product_id)
    
    # 주문 생성
    total_price = price * order.quantity
    new_order = {
        "id": 1,  # 실제로는 DB에서 생성된 ID
        "user_id": order.user_id,
        "product_id": order.product_id,
        "quantity": order.quantity,
        "total_price": total_price
    }
    
    # 실제 구현에서는 DB에 저장
    
    return new_order
      

메시지 발행자(Publisher):

# order_service_publisher.py
import pika
import json
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Order Service (Publisher)")

class OrderCreate(BaseModel):
    user_id: int
    product_id: int
    quantity: int

class Order(BaseModel):
    id: int
    user_id: int
    product_id: int
    quantity: int

# RabbitMQ 연결 설정
def get_rabbitmq_connection():
    credentials = pika.PlainCredentials('guest', 'guest')
    parameters = pika.ConnectionParameters('rabbitmq', 5672, '/', credentials)
    return pika.BlockingConnection(parameters)

@app.post("/orders/", response_model=Order)
def create_order(order: OrderCreate):
    # 주문 생성
    new_order = {
        "id": 1,  # 실제로는 DB에서 생성된 ID
        "user_id": order.user_id,
        "product_id": order.product_id,
        "quantity": order.quantity
    }
    
    # 실제 구현에서는 DB에 저장
    
    # 주문 생성 이벤트 발행
    connection = get_rabbitmq_connection()
    channel = connection.channel()
    
    # 큐 선언
    channel.queue_declare(queue='order_created')
    
    # 메시지 발행
    channel.basic_publish(
        exchange='',
        routing_key='order_created',
        body=json.dumps(new_order),
        properties=pika.BasicProperties(
            delivery_mode=2,  # 메시지 지속성 보장
        )
    )
    
    connection.close()
    
    return new_order
      

메시지 소비자(Consumer):

# inventory_service_consumer.py
import pika
import json
import time

def process_order(order):
    """주문을 처리하고 재고를 업데이트하는 함수"""
    print(f"Processing order {order['id']} for product {order['product_id']}")
    # 실제 구현에서는 데이터베이스에서 재고 업데이트
    print(f"Updated inventory for product {order['product_id']}")

def callback(ch, method, properties, body):
    """메시지를 받았을 때 실행되는 콜백 함수"""
    order = json.loads(body)
    print(f"Received order: {order}")
    
    try:
        process_order(order)
        # 메시지 처리 완료 확인
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"Error processing order: {str(e)}")
        # 처리 실패 시 메시지 다시 큐에 넣기
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

def start_consuming():
    """메시지 소비 시작"""
    # RabbitMQ 연결 설정
    credentials = pika.PlainCredentials('guest', 'guest')
    parameters = pika.ConnectionParameters('rabbitmq', 5672, '/', credentials)
    
    # 연결 시도 (재시도 로직 포함)
    connected = False
    while not connected:
        try:
            connection = pika.BlockingConnection(parameters)
            connected = True
        except pika.exceptions.AMQPConnectionError:
            print("Failed to connect to RabbitMQ. Retrying in 5 seconds...")
            time.sleep(5)
    
    channel = connection.channel()
    
    # 큐 선언
    channel.queue_declare(queue='order_created')
    
    # 한 번에 하나의 메시지만 처리하도록 설정
    channel.basic_qos(prefetch_count=1)
    
    # 콜백 함수 등록
    channel.basic_consume(queue='order_created', on_message_callback=callback)
    
    print("Inventory service started. Waiting for orders...")
    channel.start_consuming()

if __name__ == '__main__':
    start_consuming()
      

사용자 서비스 (PostgreSQL 사용):

# user_service/database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# PostgreSQL 연결 설정
DATABASE_URL = "postgresql://username:password@postgres-user-db:5432/userdb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 사용자 모델
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)
    hashed_password = Column(String)

# 데이터베이스 초기화
def init_db():
    Base.metadata.create_all(bind=engine)
      

상품 서비스 (MongoDB 사용):

# product_service/database.py
from pymongo import MongoClient
from bson.objectid import ObjectId

# MongoDB 연결 설정
client = MongoClient("mongodb://mongo-product-db:27017/")
db = client["productdb"]
products_collection = db["products"]

# 상품 CRUD 함수
def create_product(product_data):
    result = products_collection.insert_one(product_data)
    return str(result.inserted_id)

def get_product(product_id):
    return products_collection.find_one({"_id": ObjectId(product_id)})

def get_products(skip=0, limit=100):
    return list(products_collection.find().skip(skip).limit(limit))

def update_product(product_id, product_data):
    return products_collection.update_one(
        {"_id": ObjectId(product_id)},
        {"$set": product_data}
    )

def delete_product(product_id):
    return products_collection.delete_one({"_id": ObjectId(product_id)})
      

주문 서비스 (MySQL 사용):

# order_service/database.py
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import datetime

# MySQL 연결 설정
DATABASE_URL = "mysql+pymysql://username:password@mysql-order-db:3306/orderdb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 주문 모델
class Order(Base):
    __tablename__ = "orders"
    
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, index=True)
    total_amount = Column(Float)
    status = Column(String(20))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

class OrderItem(Base):
    __tablename__ = "order_items"
    
    id = Column(Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey("orders.id"))
    product_id = Column(Integer, index=True)
    quantity = Column(Integer)
    price = Column(Float)
    
    order = relationship("Order", back_populates="items")

Order.items = relationship("OrderItem", back_populates="order")

# 데이터베이스 초기화
def init_db():
    Base.metadata.create_all(bind=engine)
      

# cqrs_example.py
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime
import json
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
import asyncio

app = FastAPI(title="CQRS Example")

# 명령 모델 (쓰기 작업용)
command_engine = create_engine("postgresql://username:password@postgres-command-db:5432/commanddb")
CommandSession = sessionmaker(autocommit=False, autoflush=False, bind=command_engine)
CommandBase = declarative_base()

class ProductCommand(CommandBase):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(Text)
    price = Column(Float)
    stock = Column(Integer)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)

# 쿼리 모델 (읽기 작업용)
query_engine = create_engine("postgresql://username:password@postgres-query-db:5432/querydb")
QuerySession = sessionmaker(autocommit=False, autoflush=False, bind=query_engine)
QueryBase = declarative_base()

class ProductQuery(QueryBase):
    __tablename__ = "product_views"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(Text)
    price = Column(Float)
    stock = Column(Integer)
    category = Column(String, index=True)
    tags = Column(String)  # JSON 문자열로 저장된 태그
    average_rating = Column(Float)
    review_count = Column(Integer)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow)

# 데이터베이스 초기화
CommandBase.metadata.create_all(bind=command_engine)
QueryBase.metadata.create_all(bind=query_engine)

# 이벤트 저장소 (실제로는 메시지 큐나 이벤트 스트림을 사용)
class EventStore:
    events = []
    
    @classmethod
    def publish_event(cls, event_type, event_data):
        event = {
            "id": len(cls.events) + 1,
            "timestamp": datetime.datetime.utcnow().isoformat(),
            "type": event_type,
            "data": event_data
        }
        cls.events.append(event)
        # 실제로는 여기서 메시지 큐에 이벤트를 발행
        asyncio.create_task(cls.process_event(event))
        return event
    
    @classmethod
    async def process_event(cls, event):
        # 이벤트 처리 (쿼리 모델 업데이트)
        if event["type"] == "ProductCreated" or event["type"] == "ProductUpdated":
            data = event["data"]
            db = QuerySession()
            try:
                # 쿼리 모델에서 제품 찾기 또는 새로 생성
                product = db.query(ProductQuery).filter(ProductQuery.id == data["id"]).first()
                if not product:
                    product = ProductQuery(id=data["id"])
                
                # 기본 필드 업데이트
                product.name = data["name"]
                product.description = data["description"]
                product.price = data["price"]
                product.stock = data["stock"]
                
                # 추가 필드 (실제로는 다른 서비스에서 가져올 수 있음)
                if "category" in data:
                    product.category = data["category"]
                if "tags" in data:
                    product.tags = json.dumps(data["tags"])
                if "average_rating" in data:
                    product.average_rating = data["average_rating"]
                if "review_count" in data:
                    product.review_count = data["review_count"]
                
                product.updated_at = datetime.datetime.utcnow()
                
                db.add(product)
                db.commit()
                print(f"Query model updated for product {data['id']}")
            except Exception as e:
                db.rollback()
                print(f"Error updating query model: {str(e)}")
            finally:
                db.close()

# API 모델
class ProductCreate(BaseModel):
    name: str
    description: str
    price: float
    stock: int
    category: str = None
    tags: list = None

class ProductResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    stock: int
    category: str = None
    tags: list = None
    average_rating: float = None
    review_count: int = None

# 의존성
def get_command_db():
    db = CommandSession()
    try:
        yield db
    finally:
        db.close()

def get_query_db():
    db = QuerySession()
    try:
        yield db
    finally:
        db.close()

# 명령 API (쓰기 작업)
@app.post("/products/", response_model=ProductResponse)
def create_product(product: ProductCreate, db: CommandSession = Depends(get_command_db)):
    db_product = ProductCommand(
        name=product.name,
        description=product.description,
        price=product.price,
        stock=product.stock
    )
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    
    # 이벤트 발행
    event_data = {
        "id": db_product.id,
        "name": db_product.name,
        "description": db_product.description,
        "price": db_product.price,
        "stock": db_product.stock,
        "category": product.category,
        "tags": product.tags
    }
    EventStore.publish_event("ProductCreated", event_data)
    
    return db_product

@app.put("/products/{product_id}", response_model=ProductResponse)
def update_product(product_id: int, product: ProductCreate, db: CommandSession = Depends(get_command_db)):
    db_product = db.query(ProductCommand).filter(ProductCommand.id == product_id).first()
    if db_product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    
    db_product.name = product.name
    db_product.description = product.description
    db_product.price = product.price
    db_product.stock = product.stock
    db_product.updated_at = datetime.datetime.utcnow()
    
    db.commit()
    db.refresh(db_product)
    
    # 이벤트 발행
    event_data = {
        "id": db_product.id,
        "name": db_product.name,
        "description": db_product.description,
        "price": db_product.price,
        "stock": db_product.stock,
        "category": product.category,
        "tags": product.tags
    }
    EventStore.publish_event("ProductUpdated", event_data)
    
    return db_product

# 쿼리 API (읽기 작업)
@app.get("/products/{product_id}", response_model=ProductResponse)
def read_product(product_id: int, db: QuerySession = Depends(get_query_db)):
    db_product = db.query(ProductQuery).filter(ProductQuery.id == product_id).first()
    if db_product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    
    response = {
        "id": db_product.id,
        "name": db_product.name,
        "description": db_product.description,
        "price": db_product.price,
        "stock": db_product.stock,
        "category": db_product.category,
        "tags": json.loads(db_product.tags) if db_product.tags else None,
        "average_rating": db_product.average_rating,
        "review_count": db_product.review_count
    }
    
    return response

@app.get("/products/", response_model=list[ProductResponse])
def read_products(skip: int = 0, limit: int = 100, category: str = None, db: QuerySession = Depends(get_query_db)):
    query = db.query(ProductQuery)
    
    if category:
        query = query.filter(ProductQuery.category == category)
    
    products = query.offset(skip).limit(limit).all()
    
    response = []
    for db_product in products:
        product_data = {
            "id": db_product.id,
            "name": db_product.name,
            "description": db_product.description,
            "price": db_product.price,
            "stock": db_product.stock,
            "category": db_product.category,
            "tags": json.loads(db_product.tags) if db_product.tags else None,
            "average_rating": db_product.average_rating,
            "review_count": db_product.review_count
        }
        response.append(product_data)
    
    return response
      

먼저 각 서비스별로 Dockerfile을 만들어야 해. 사용자 서비스의 Dockerfile 예시:

# user-service/Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 의존성 파일 복사 및 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 포트 노출
EXPOSE 8000

# 헬스체크 엔드포인트 추가
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

# 애플리케이션 실행
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
      

requirements.txt 파일:

fastapi==0.110.0
uvicorn==0.27.1
sqlalchemy==2.0.27
pydantic==2.5.3
psycopg2-binary==2.9.9
python-jose==3.3.0
passlib==1.7.4
python-multipart==0.0.9
      

이제 여러 서비스를 함께 실행하기 위한 docker-compose.yml 파일을 만들자:

# docker-compose.yml
version: '3.8'

services:
  # API 게이트웨이
  api-gateway:
    build: ./api-gateway
    ports:
      - "8080:8080"
    depends_on:
      - user-service
      - product-service
      - order-service
    environment:
      - USER_SERVICE_URL=http://user-service:8000
      - PRODUCT_SERVICE_URL=http://product-service:8001
      - ORDER_SERVICE_URL=http://order-service:8002
    networks:
      - microservices-net

  # 사용자 서비스
  user-service:
    build: ./user-service
    ports:
      - "8000:8000"
    depends_on:
      - postgres-user-db
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@postgres-user-db:5432/userdb
      - SECRET_KEY=your-secret-key
    networks:
      - microservices-net

  # 상품 서비스
  product-service:
    build: ./product-service
    ports:
      - "8001:8001"
    depends_on:
      - mongo-product-db
    environment:
      - MONGODB_URI=mongodb://mongo-product-db:27017/productdb
    networks:
      - microservices-net

  # 주문 서비스
  order-service:
    build: ./order-service
    ports:
      - "8002:8002"
    depends_on:
      - mysql-order-db
      - rabbitmq
    environment:
      - DATABASE_URL=mysql+pymysql://root:root@mysql-order-db:3306/orderdb
      - RABBITMQ_URL=amqp://guest:guest@rabbitmq:5672/
    networks:
      - microservices-net

  # 사용자 데이터베이스
  postgres-user-db:
    image: postgres:15
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=userdb
    volumes:
      - postgres-user-data:/var/lib/postgresql/data
    networks:
      - microservices-net

  # 상품 데이터베이스
  mongo-product-db:
    image: mongo:6
    volumes:
      - mongo-product-data:/data/db
    networks:
      - microservices-net

  # 주문 데이터베이스
  mysql-order-db:
    image: mysql:8
    environment:
      - MYSQL_ROOT_PASSWORD=root
      - MYSQL_DATABASE=orderdb
    volumes:
      - mysql-order-data:/var/lib/mysql
    networks:
      - microservices-net

  # 메시지 브로커
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    networks:
      - microservices-net

networks:
  microservices-net:
    driver: bridge

volumes:
  postgres-user-data:
  mongo-product-data:
  mysql-order-data:
      

사용자 서비스의 Deployment 및 Service:

# user-service-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  labels:
    app: user-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: user-service
  template:
    metadata:
      labels:
        app: user-service
    spec:
      containers:
      - name: user-service
        image: your-registry/user-service:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: user-service-secrets
              key: database-url
        - name: SECRET_KEY
          valueFrom:
            secretKeyRef:
              name: user-service-secrets
              key: secret-key
        resources:
          limits:
            cpu: "500m"
            memory: "512Mi"
          requests:
            cpu: "100m"
            memory: "256Mi"
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 15
          periodSeconds: 20
---
apiVersion: v1
kind: Service
metadata:
  name: user-service
spec:
  selector:
    app: user-service
  ports:
  - port: 8000
    targetPort: 8000
  type: ClusterIP
      

API 게이트웨이의 Deployment 및 Service:

# api-gateway-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-gateway
  labels:
    app: api-gateway
spec:
  replicas: 2
  selector:
    matchLabels:
      app: api-gateway
  template:
    metadata:
      labels:
        app: api-gateway
    spec:
      containers:
      - name: api-gateway
        image: your-registry/api-gateway:latest
        ports:
        - containerPort: 8080
        env:
        - name: USER_SERVICE_URL
          value: "http://user-service:8000"
        - name: PRODUCT_SERVICE_URL
          value: "http://product-service:8001"
        - name: ORDER_SERVICE_URL
          value: "http://order-service:8002"
        resources:
          limits:
            cpu: "500m"
            memory: "512Mi"
          requests:
            cpu: "100m"
            memory: "256Mi"
---
apiVersion: v1
kind: Service
metadata:
  name: api-gateway
spec:
  selector:
    app: api-gateway
  ports:
  - port: 80
    targetPort: 8080
  type: LoadBalancer
      

Kubernetes ConfigMap 및 Secret:

# configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: microservices-config
data:
  ENVIRONMENT: "production"
  LOG_LEVEL: "info"
---
# secrets.yaml (실제로는 base64로 인코딩된 값을 사용)
apiVersion: v1
kind: Secret
metadata:
  name: user-service-secrets
type: Opaque
data:
  database-url: cG9zdGdyZXNxbDovL3Bvc3RncmVzOnBvc3RncmVzQHBvc3RncmVzLXVzZXItZGI6NTQzMi91c2VyZGI=
  secret-key: eW91ci1zZWNyZXQta2V5
      

# .github/workflows/user-service-ci-cd.yml
name: User Service CI/CD

on:
  push:
    branches: [ main ]
    paths:
      - 'user-service/**'
      - '.github/workflows/user-service-ci-cd.yml'
  pull_request:
    branches: [ main ]
    paths:
      - 'user-service/**'

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    - name: Install dependencies
      run: |
        cd user-service
        python -m pip install --upgrade pip
        pip install pytest pytest-cov
        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
    - name: Test with pytest
      run: |
        cd user-service
        pytest --cov=. --cov-report=xml
    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v3
      with:
        file: ./user-service/coverage.xml
        flags: user-service
        fail_ci_if_error: true

  build-and-push:
    needs: test
    runs-on: ubuntu-latest
    if: github.event_name == 'push'
    steps:
    - uses: actions/checkout@v3
    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v2
    - name: Login to DockerHub
      uses: docker/login-action@v2
      with:
        username: ${{ secrets.DOCKERHUB_USERNAME }}
        password: ${{ secrets.DOCKERHUB_TOKEN }}
    - name: Build and push
      uses: docker/build-push-action@v4
      with:
        context: ./user-service
        push: true
        tags: your-registry/user-service:latest,your-registry/user-service:${{ github.sha }}
        cache-from: type=registry,ref=your-registry/user-service:buildcache
        cache-to: type=registry,ref=your-registry/user-service:buildcache,mode=max

  deploy:
    needs: build-and-push
    runs-on: ubuntu-latest
    if: github.event_name == 'push'
    steps:
    - uses: actions/checkout@v3
    - name: Set up kubectl
      uses: azure/setup-kubectl@v3
    - name: Set Kubernetes context
      uses: azure/k8s-set-context@v3
      with:
        kubeconfig: ${{ secrets.KUBE_CONFIG }}
    - name: Update deployment image
      run: |
        kubectl set image deployment/user-service user-service=your-registry/user-service:${{ github.sha }} --record
    - name: Verify deployment
      run: |
        kubectl rollout status deployment/user-service
      

# logger.py
import logging
import json
import uuid
import time
from datetime import datetime
import socket
import os
from pythonjsonlogger import jsonlogger

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
        
        # 표준 필드 추가
        log_record['timestamp'] = datetime.utcnow().isoformat()
        log_record['level'] = record.levelname
        log_record['logger'] = record.name
        
        # 서비스 정보 추가
        log_record['service'] = os.environ.get('SERVICE_NAME', 'unknown-service')
        log_record['host'] = socket.gethostname()
        log_record['environment'] = os.environ.get('ENVIRONMENT', 'development')
        
        # 요청 추적을 위한 필드
        if hasattr(record, 'request_id'):
            log_record['request_id'] = record.request_id
        else:
            log_record['request_id'] = 'no-request-id'

def get_logger(name):
    logger = logging.getLogger(name)
    
    # 로거가 이미 핸들러를 가지고 있으면 중복 설정 방지
    if logger.handlers:
        return logger
    
    # 로그 레벨 설정
    log_level = os.environ.get('LOG_LEVEL', 'INFO').upper()
    logger.setLevel(getattr(logging, log_level))
    
    # 콘솔 핸들러 추가
    handler = logging.StreamHandler()
    
    # JSON 포맷터 설정
    formatter = CustomJsonFormatter('%(timestamp)s %(level)s %(name)s %(message)s')
    handler.setFormatter(formatter)
    
    logger.addHandler(handler)
    return logger

# 미들웨어에서 사용할 요청 ID 생성 함수
def generate_request_id():
    return str(uuid.uuid4())

# FastAPI 미들웨어 예시
class RequestIdMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        
        # 요청 ID 생성
        request_id = generate_request_id()
        
        # 로깅 컨텍스트에 요청 ID 추가
        logging.LoggerAdapter(get_logger("request"), {"request_id": request_id})
        
        # 스코프에 요청 ID 추가 (미들웨어에서 접근 가능)
        scope["request_id"] = request_id
        
        # 응답 헤더에 요청 ID 추가
        original_send = send
        
        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                headers = message.get("headers", [])
                headers.append((b"X-Request-ID", request_id.encode()))
                message["headers"] = headers
            await original_send(message)
        
        await self.app(scope, receive, send_wrapper)

# 사용 예시
logger = get_logger(__name__)

def process_order(order_id, user_id):
    logger.info("Processing order", extra={
        "order_id": order_id,
        "user_id": user_id,
        "operation": "process_order"
    })
    
    try:
        # 주문 처리 로직
        logger.debug("Order details retrieved", extra={
            "order_id": order_id,
            "status": "processing"
        })
        
        # 결제 처리
        logger.info("Payment processed", extra={
            "order_id": order_id,
            "payment_status": "success"
        })
        
        return True
    except Exception as e:
        logger.error("Order processing failed", extra={
            "order_id": order_id,
            "error": str(e),
            "exception_type": type(e).__name__
        })
        return False
      

# metrics.py
from prometheus_client import Counter, Histogram, Gauge, Summary, start_http_server
import time
import random
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

# 메트릭 정의
REQUEST_COUNT = Counter(
    'app_request_count',
    'Application Request Count',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'app_request_latency_seconds',
    'Application Request Latency',
    ['method', 'endpoint']
)

ACTIVE_REQUESTS = Gauge(
    'app_active_requests',
    'Active Requests'
)

DB_QUERY_LATENCY = Summary(
    'app_db_query_latency_seconds',
    'Database Query Latency'
)

# FastAPI 미들웨어
class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        # 활성 요청 수 증가
        ACTIVE_REQUESTS.inc()
        
        # 메서드와 경로 추출
        method = request.method
        endpoint = request.url.path
        
        try:
            response = await call_next(request)
            
            # 요청 카운터 증가
            REQUEST_COUNT.labels(
                method=method,
                endpoint=endpoint,
                status=response.status_code
            ).inc()
            
            return response
        except Exception as e:
            # 예외 발생 시 500 상태 코드로 카운트
            REQUEST_COUNT.labels(
                method=method,
                endpoint=endpoint,
                status=500
            ).inc()
            raise e
        finally:
            # 요청 처리 시간 기록
            REQUEST_LATENCY.labels(
                method=method,
                endpoint=endpoint
            ).observe(time.time() - start_time)
            
            # 활성 요청 수 감소
            ACTIVE_REQUESTS.dec()

# 데이터베이스 쿼리 시간 측정 데코레이터
def track_db_query_time(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        DB_QUERY_LATENCY.observe(time.time() - start_time)
        return result
    return wrapper

# FastAPI 애플리케이션에 미들웨어 등록 예시
def setup_metrics(app: FastAPI, port=8000):
    # Prometheus 메트릭 서버 시작
    start_http_server(port)
    
    # 미들웨어 등록
    app.add_middleware(PrometheusMiddleware)
    
    # 메트릭 엔드포인트 추가
    @app.get("/metrics")
    async def metrics():
        return {"message": f"Metrics available at http://localhost:{port}"}

# 사용 예시
app = FastAPI()
setup_metrics(app, port=8001)  # 메트릭은 별도 포트에서 제공

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 데이터베이스 쿼리 시뮬레이션
    query_result = simulate_db_query(user_id)
    return {"user_id": user_id, "data": query_result}

@track_db_query_time
def simulate_db_query(user_id):
    # 데이터베이스 쿼리 시뮬레이션
    time.sleep(random.uniform(0.05, 0.2))  # 50-200ms 쿼리 시간 시뮬레이션
    return {"name": f"User {user_id}", "email": f"user{user_id}@example.com"}
      

# tracing.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from fastapi import FastAPI
import os

def setup_tracing(app: FastAPI, service_name: str):
    # 리소스 설정
    resource = Resource(attributes={
        SERVICE_NAME: service_name
    })
    
    # 트레이서 프로바이더 설정
    tracer_provider = TracerProvider(resource=resource)
    
    # OTLP 익스포터 설정 (Jaeger 또는 다른 백엔드로 전송)
    otlp_exporter = OTLPSpanExporter(
        endpoint=os.environ.get("OTLP_ENDPOINT", "http://jaeger:4317")
    )
    
    # 배치 프로세서 설정
    span_processor = BatchSpanProcessor(otlp_exporter)
    tracer_provider.add_span_processor(span_processor)
    
    # 글로벌 트레이서 프로바이더 설정
    trace.set_tracer_provider(tracer_provider)
    
    # FastAPI 계측
    FastAPIInstrumentor.instrument_app(app)
    
    # HTTP 클라이언트 계측
    RequestsInstrumentor().instrument()
    
    # SQLAlchemy 계측 (선택적)
    # SQLAlchemyInstrumentor().instrument()
    
    return trace.get_tracer(service_name)

# 사용 예시
app = FastAPI()
tracer = setup_tracing(app, "user-service")

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 수동 트레이싱 예시
    with tracer.start_as_current_span("get_user_details") as span:
        # 스팬에 속성 추가
        span.set_attribute("user.id", user_id)
        
        # 데이터베이스 쿼리 시뮬레이션
        user_data = await get_user_from_database(user_id)
        
        # 결과에 따라 스팬 상태 설정
        if user_data:
            span.set_status(trace.Status(trace.StatusCode.OK))
        else:
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            span.record_exception(Exception(f"User {user_id} not found"))
        
        return user_data

async def get_user_from_database(user_id: int):
    # 중첩된 스팬 생성
    with tracer.start_as_current_span("database.query") as span:
        span.set_attribute("db.query", f"SELECT * FROM users WHERE id = {user_id}")
        
        # 데이터베이스 쿼리 시뮬레이션
        import time
        time.sleep(0.1)
        
        # 다른 서비스 호출 시뮬레이션
        await get_user_preferences(user_id)
        
        return {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}

async def get_user_preferences(user_id: int):
    # 다른 서비스 호출을 나타내는 스팬
    with tracer.start_as_current_span("get_user_preferences") as span:
        span.set_attribute("user.id", user_id)
        
        # 외부 서비스 호출 시뮬레이션
        import time
        time.sleep(0.05)
        
        return {"theme": "dark", "language": "en"}
      

# user_service/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
import os

from database import SessionLocal, engine, Base
from models import User as UserModel
import schemas

# 데이터베이스 초기화
Base.metadata.create_all(bind=engine)

app = FastAPI(title="User Service")

# 보안 설정
SECRET_KEY = os.environ.get("SECRET_KEY", "your-secret-key")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 유틸리티 함수
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    return db.query(UserModel).filter(UserModel.username == username).first()

def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(db, username=username)
    if user is None:
        raise credentials_exception
    return user

# API 엔드포인트
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = get_user(db, username=user.username)
    if db_user:
        raise HTTPException(status_code=400, detail="Username already registered")
    hashed_password = get_password_hash(user.password)
    db_user = UserModel(
        username=user.username,
        email=user.email,
        full_name=user.full_name,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.post("/token", response_model=schemas.Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(current_user: UserModel = Depends(get_current_user)):
    return current_user

@app.put("/users/me/", response_model=schemas.User)
async def update_user(
    user_update: schemas.UserUpdate,
    current_user: UserModel = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    if user_update.email:
        current_user.email = user_update.email
    if user_update.full_name:
        current_user.full_name = user_update.full_name
    if user_update.password:
        current_user.hashed_password = get_password_hash(user_update.password)
    
    db.commit()
    db.refresh(current_user)
    return current_user

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

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

데이터베이스 모델 (models.py):

# user_service/models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
      

스키마 정의 (schemas.py):

# user_service/schemas.py
from pydantic import BaseModel, EmailStr
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    full_name: Optional[str] = None
    password: Optional[str] = None

class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    
    class Config:
        orm_mode = True

class Token(BaseModel):
    access_token: str
    token_type: str
      

데이터베이스 설정 (database.py):

# user_service/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/userdb")

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
      

# order_service/main.py
from fastapi import FastAPI, Depends, HTTPException, Header
from sqlalchemy.orm import Session
from typing import List, Optional
import httpx
import json
import pika
import uuid
from datetime import datetime

from database import SessionLocal, engine, Base
from models import Order as OrderModel, OrderItem as OrderItemModel
import schemas

# 데이터베이스 초기화
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Order Service")

# RabbitMQ 연결 설정
def get_rabbitmq_connection():
    credentials = pika.PlainCredentials('guest', 'guest')
    parameters = pika.ConnectionParameters('rabbitmq', 5672, '/', credentials)
    return pika.BlockingConnection(parameters)

# 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user_id(authorization: Optional[str] = Header(None)):
    if authorization is None:
        raise HTTPException(status_code=401, detail="Authorization header missing")
    
    # 실제 구현에서는 API 게이트웨이에서 전달된 사용자 ID를 추출
    # 여기서는 간단히 헤더에서 사용자 ID를 추출한다고 가정
    try:
        token = authorization.split(" ")[1]
        # 실제로는 토큰 검증 로직이 필요
        user_id = 1  # 테스트용 사용자 ID
        return user_id
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid token")

# 상품 서비스 API 호출
async def get_product(product_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://product-service:8001/products/{product_id}")
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code, detail="Product service error")
        return response.json()

# API 엔드포인트
@app.post("/cart/items/", response_model=schemas.CartItem)
async def add_to_cart(
    item: schemas.CartItemCreate,
    user_id: int = Depends(get_current_user_id),
    db: Session = Depends(get_db)
):
    # 상품 정보 가져오기
    product = await get_product(item.product_id)
    
    # 장바구니에 상품 추가 (실제로는 DB에 저장)
    cart_item = {
        "id": str(uuid.uuid4()),
        "user_id": user_id,
        "product_id": item.product_id,
        "product_name": product["name"],
        "quantity": item.quantity,
        "price": product["price"]
    }
    
    # 실제 구현에서는 DB에 저장
    
    return cart_item

@app.get("/cart/", response_model=List[schemas.CartItem])
async def get_cart(
    user_id: int = Depends(get_current_user_id)
):
    # 실제 구현에서는 DB에서 사용자의 장바구니 항목을 가져옴
    # 여기서는 테스트 데이터 반환
    return [
        {
            "id": "1",
            "user_id": user_id,
            "product_id": 1,
            "product_name": "Laptop",
            "quantity": 1,
            "price": 1299.99
        },
        {
            "id": "2",
            "user_id": user_id,
            "product_id": 2,
            "product_name": "Smartphone",
            "quantity": 2,
            "price": 699.99
        }
    ]

@app.post("/orders/", response_model=schemas.Order)
async def create_order(
    order_create: schemas.OrderCreate,
    user_id: int = Depends(get_current_user_id),
    db: Session = Depends(get_db)
):
    # 주문 생성
    db_order = OrderModel(
        user_id=user_id,
        status="pending",
        shipping_address=order_create.shipping_address,
        total_amount=0  # 초기값, 나중에 계산
    )
    db.add(db_order)
    db.commit()
    db.refresh(db_order)
    
    total_amount = 0
    
    # 주문 항목 추가
    for item in order_create.items:
        # 상품 정보 가져오기
        product = await get_product(item.product_id)
        
        # 주문 항목 생성
        db_order_item = OrderItemModel(
            order_id=db_order.id,
            product_id=item.product_id,
            quantity=item.quantity,
            price=product["price"]
        )
        db.add(db_order_item)
        
        # 총액 계산
        total_amount += product["price"] * item.quantity
    
    # 주문 총액 업데이트
    db_order.total_amount = total_amount
    db.commit()
    db.refresh(db_order)
    
    # 주문 생성 이벤트 발행
    connection = get_rabbitmq_connection()
    channel = connection.channel()
    
    # 큐 선언
    channel.queue_declare(queue='order_created')
    
    # 메시지 발행
    order_data = {
        "order_id": db_order.id,
        "user_id": user_id,
        "total_amount": total_amount,
        "status": "pending",
        "created_at": datetime.utcnow().isoformat()
    }
    
    channel.basic_publish(
        exchange='',
        routing_key='order_created',
        body=json.dumps(order_data),
        properties=pika.BasicProperties(
            delivery_mode=2,  # 메시지 지속성 보장
        )
    )
    
    connection.close()
    
    return db_order

@app.get("/orders/{order_id}", response_model=schemas.OrderDetail)
async def get_order(
    order_id: int,
    user_id: int = Depends(get_current_user_id),
    db: Session = Depends(get_db)
):
    # 주문 정보 가져오기
    db_order = db.query(OrderModel).filter(OrderModel.id == order_id, OrderModel.user_id == user_id).first()
    if db_order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    
    # 주문 항목 가져오기
    db_order_items = db.query(OrderItemModel).filter(OrderItemModel.order_id == order_id).all()
    
    # 응답 구성
    order_detail = {
        "id": db_order.id,
        "user_id": db_order.user_id,
        "status": db_order.status,
        "shipping_address": db_order.shipping_address,
        "total_amount": db_order.total_amount,
        "created_at": db_order.created_at,
        "items": []
    }
    
    for item in db_order_items:
        order_detail["items"].append({
            "id": item.id,
            "product_id": item.product_id,
            "quantity": item.quantity,
            "price": item.price
        })
    
    return order_detail

@app.get("/orders/", response_model=List[schemas.Order])
async def get_orders(
    user_id: int = Depends(get_current_user_id),
    db: Session = Depends(get_db)
):
    # 사용자의 모든 주문 가져오기
    db_orders = db.query(OrderModel).filter(OrderModel.user_id == user_id).all()
    return db_orders

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8002, reload=True)
      

# caching.py
import redis
import json
import functools
import hashlib
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import time

# Redis 연결 설정
redis_client = redis.Redis(host='redis', port=6379, db=0)

# 캐시 데코레이터
def cache(expire=3600):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # 캐시 키 생성
            key_parts = [func.__name__]
            key_parts.extend([str(arg) for arg in args])
            key_parts.extend([f"{k}:{v}" for k, v in sorted(kwargs.items())])
            cache_key = hashlib.md5(":".join(key_parts).encode()).hexdigest()
            
            # 캐시에서 데이터 조회
            cached_data = redis_client.get(cache_key)
            if cached_data:
                return json.loads(cached_data)
            
            # 캐시에 없으면 함수 실행
            result = await func(*args, **kwargs)
            
            # 결과를 캐시에 저장
            redis_client.setex(
                cache_key,
                expire,
                json.dumps(result, default=str)  # datetime 등을 문자열로 변환
            )
            
            return result
        return wrapper
    return decorator

# FastAPI 애플리케이션에 적용
app = FastAPI(title="Product Service with Caching")

# 데이터베이스 의존성 (예시)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 캐시를 사용하는 API 엔드포인트
@app.get("/products/{product_id}")
@cache(expire=300)  # 5분 캐시
async def get_product(product_id: int, db: Session = Depends(get_db)):
    # 데이터베이스 쿼리 시뮬레이션
    time.sleep(0.5)  # 실제로는 DB 쿼리
    
    # 실제 구현에서는 DB에서 제품 정보를 가져옴
    product = {
        "id": product_id,
        "name": f"Product {product_id}",
        "description": "A high-quality product",
        "price": 99.99,
        "stock": 100
    }
    
    return product

# 캐시 무효화 예시
@app.put("/products/{product_id}")
async def update_product(product_id: int, db: Session = Depends(get_db)):
    # 제품 업데이트 로직
    
    # 캐시 무효화
    cache_key = hashlib.md5(f"get_product:{product_id}").hexdigest()
    redis_client.delete(cache_key)
    
    return {"message": "Product updated and cache invalidated"}

# 캐시 통계 엔드포인트
@app.get("/cache/stats")
async def get_cache_stats():
    info = redis_client.info()
    return {
        "used_memory": info["used_memory_human"],
        "hits": info["keyspace_hits"],
        "misses": info["keyspace_misses"],
        "hit_rate": info["keyspace_hits"] / (info["keyspace_hits"] + info["keyspace_misses"] + 0.001) * 100
    }
      

# async_example.py
from fastapi import FastAPI
import asyncio
import httpx
from typing import List

app = FastAPI(title="Async API Example")

# 비동기 HTTP 클라이언트
async def fetch_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# 순차적 API 호출 (느림)
@app.get("/sequential")
async def get_data_sequential():
    start_time = asyncio.get_event_loop().time()
    
    # 순차적으로 API 호출
    data1 = await fetch_data("https://jsonplaceholder.typicode.com/posts/1")
    data2 = await fetch_data("https://jsonplaceholder.typicode.com/posts/2")
    data3 = await fetch_data("https://jsonplaceholder.typicode.com/posts/3")
    
    end_time = asyncio.get_event_loop().time()
    execution_time = end_time - start_time
    
    return {
        "execution_time": execution_time,
        "data": [data1, data2, data3]
    }

# 병렬 API 호출 (빠름)
@app.get("/parallel")
async def get_data_parallel():
    start_time = asyncio.get_event_loop().time()
    
    # 병렬로 API 호출
    tasks = [
        fetch_data("https://jsonplaceholder.typicode.com/posts/1"),
        fetch_data("https://jsonplaceholder.typicode.com/posts/2"),
        fetch_data("https://jsonplaceholder.typicode.com/posts/3")
    ]
    
    results = await asyncio.gather(*tasks)
    
    end_time = asyncio.get_event_loop().time()
    execution_time = end_time - start_time
    
    return {
        "execution_time": execution_time,
        "data": results
    }

# 비동기 스트리밍 응답
@app.get("/stream")
async def stream_data():
    async def generate_data():
        for i in range(10):
            await asyncio.sleep(0.5)  # 비동기 대기
            yield f"data: {i}\n\n"
    
    return generate_data()
      

# database_optimization.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, func, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, joinedload, Session
from typing import List
import time

# 데이터베이스 설정
DATABASE_URL = "postgresql://username:password@postgres:5432/productdb"
engine = create_engine(DATABASE_URL, pool_size=20, max_overflow=0)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 모델 정의
class Category(Base):
    __tablename__ = "categories"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    
    products = relationship("Product", back_populates="category")

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Float)
    stock = Column(Integer)
    category_id = Column(Integer, ForeignKey("categories.id"))
    
    category = relationship("Category", back_populates="products")
    reviews = relationship("Review", back_populates="product")
    
    # 복합 인덱스 추가
    __table_args__ = (
        Index('idx_product_name_price', 'name', 'price'),
    )

class Review(Base):
    __tablename__ = "reviews"
    
    id = Column(Integer, primary_key=True, index=True)
    product_id = Column(Integer, ForeignKey("products.id"), index=True)
    rating = Column(Integer)
    comment = Column(String)
    
    product = relationship("Product", back_populates="reviews")

# 데이터베이스 초기화
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Database Optimization Example")

# 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 최적화되지 않은 쿼리
@app.get("/products/unoptimized")
def get_products_unoptimized(db: Session = Depends(get_db)):
    start_time = time.time()
    
    # 모든 제품 가져오기
    products = db.query(Product).all()
    
    # N+1 문제: 각 제품에 대해 별도의 쿼리로 카테고리와 리뷰 가져오기
    result = []
    for product in products:
        # 이 접근은 각 제품마다 추가 쿼리를 발생시킴
        category = product.category
        reviews = product.reviews
        
        avg_rating = sum(review.rating for review in reviews) / len(reviews) if reviews else 0
        
        result.append({
            "id": product.id,
            "name": product.name,
            "price": product.price,
            "category": category.name,
            "avg_rating": avg_rating,
            "review_count": len(reviews)
        })
    
    execution_time = time.time() - start_time
    
    return {
        "execution_time": execution_time,
        "count": len(result),
        "products": result[:10]  # 처음 10개만 반환
    }

# 최적화된 쿼리
@app.get("/products/optimized")
def get_products_optimized(db: Session = Depends(get_db)):
    start_time = time.time()
    
    # 조인을 사용하여 한 번의 쿼리로 제품, 카테고리, 리뷰 가져오기
    products = db.query(Product).options(
        joinedload(Product.category),
        joinedload(Product.reviews)
    ).all()
    
    result = []
    for product in products:
        avg_rating = sum(review.rating for review in product.reviews) / len(product.reviews) if product.reviews else 0
        
        result.append({
            "id": product.id,
            "name": product.name,
            "price": product.price,
            "category": product.category.name,
            "avg_rating": avg_rating,
            "review_count": len(product.reviews)
        })
    
    execution_time = time.time() - start_time
    
    return {
        "execution_time": execution_time,
        "count": len(result),
        "products": result[:10]  # 처음 10개만 반환
    }

# 더 최적화된 쿼리 (집계 함수 사용)
@app.get("/products/highly-optimized")
def get_products_highly_optimized(db: Session = Depends(get_db)):
    start_time = time.time()
    
    # 서브쿼리를 사용하여 리뷰 집계
    review_stats = db.query(
        Review.product_id,
        func.avg(Review.rating).label("avg_rating"),
        func.count(Review.id).label("review_count")
    ).group_by(Review.product_id).subquery()
    
    # 조인을 사용하여 한 번의 쿼리로 모든 데이터 가져오기
    products = db.query(
        Product,
        Category.name.label("category_name"),
        review_stats.c.avg_rating,
        review_stats.c.review_count
    ).join(
        Category, Product.category_id == Category.id
    ).outerjoin(
        review_stats, Product.id == review_stats.c.product_id
    ).all()
    
    result = []
    for product, category_name, avg_rating, review_count in products:
        result.append({
            "id": product.id,
            "name": product.name,
            "price": product.price,
            "category": category_name,
            "avg_rating": float(avg_rating) if avg_rating else 0,
            "review_count": review_count or 0
        })
    
    execution_time = time.time() - start_time
    
    return {
        "execution_time": execution_time,
        "count": len(result),
        "products": result[:10]  # 처음 10개만 반환
    }

# 페이지네이션 적용
@app.get("/products/paginated")
def get_products_paginated(page: int = 1, page_size: int = 20, db: Session = Depends(get_db)):
    start_time = time.time()
    
    # 오프셋 계산
    offset = (page - 1) * page_size
    
    # 전체 개수 가져오기
    total_count = db.query(func.count(Product.id)).scalar()
    
    # 페이지네이션 적용하여 쿼리
    products = db.query(
        Product,
        Category.name.label("category_name")
    ).join(
        Category, Product.category_id == Category.id
    ).offset(offset).limit(page_size).all()
    
    result = []
    for product, category_name in products:
        result.append({
            "id": product.id,
            "name": product.name,
            "price": product.price,
            "category": category_name
        })
    
    execution_time = time.time() - start_time
    
    return {
        "execution_time": execution_time,
        "page": page,
        "page_size": page_size,
        "total_count": total_count,
        "total_pages": (total_count + page_size - 1) // page_size,
        "products": result
    }
      

# serverless_example.py
import json
import boto3
from datetime import datetime

# DynamoDB 클라이언트
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Products')

def lambda_handler(event, context):
    """
    AWS Lambda 핸들러 함수
    """
    # API Gateway에서 전달된 HTTP 메서드 확인
    http_method = event['httpMethod']
    
    # 경로 파라미터 확인
    path_parameters = event.get('pathParameters', {})
    product_id = path_parameters.get('product_id') if path_parameters else None
    
    # 쿼리 파라미터 확인
    query_parameters = event.get('queryStringParameters', {}) or {}
    
    # HTTP 메서드에 따라 처리
    if http_method == 'GET':
        if product_id:
            return get_product(product_id)
        else:
            return list_products(query_parameters)
    elif http_method == 'POST':
        return create_product(json.loads(event['body']))
    elif http_method == 'PUT':
        return update_product(product_id, json.loads(event['body']))
    elif http_method == 'DELETE':
        return delete_product(product_id)
    else:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Unsupported HTTP method'})
        }

def get_product(product_id):
    """
    제품 정보 조회
    """
    try:
        response = table.get_item(Key={'id': product_id})
        item = response.get('Item')
        
        if not item:
            return {
                'statusCode': 404,
                'body': json.dumps({'error': 'Product not found'})
            }
        
        return {
            'statusCode': 200,
            'body': json.dumps(item)
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def list_products(query_params):
    """
    제품 목록 조회
    """
    try:
        # 페이지네이션 파라미터
        limit = int(query_params.get('limit', 20))
        last_evaluated_key = query_params.get('next_token')
        
        # 필터링 파라미터
        category = query_params.get('category')
        
        # 기본 쿼리 파라미터
        params = {
            'Limit': limit
        }
        
        # 페이지네이션 토큰이 있으면 추가
        if last_evaluated_key:
            params['ExclusiveStartKey'] = {'id': last_evaluated_key}
        
        # 카테고리 필터가 있으면 필터 표현식 추가
        if category:
            params['FilterExpression'] = 'category = :category'
            params['ExpressionAttributeValues'] = {':category': category}
        
        # 쿼리 실행
        response = table.scan(**params)
        
        result = {
            'items': response['Items'],
            'count': response['Count']
        }
        
        # 다음 페이지 토큰이 있으면 추가
        if 'LastEvaluatedKey' in response:
            result['next_token'] = response['LastEvaluatedKey']['id']
        
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def create_product(product_data):
    """
    새 제품 생성
    """
    try:
        # 제품 ID 생성 (실제로는 UUID 사용)
        product_id = str(hash(datetime.now().isoformat()))
        
        # 제품 데이터 준비
        item = {
            'id': product_id,
            'name': product_data['name'],
            'description': product_data.get('description', ''),
            'price': product_data['price'],
            'stock': product_data.get('stock', 0),
            'category': product_data.get('category', 'uncategorized'),
            'created_at': datetime.now().isoformat()
        }
        
        # DynamoDB에 저장
        table.put_item(Item=item)
        
        return {
            'statusCode': 201,
            'body': json.dumps(item)
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def update_product(product_id, product_data):
    """
    제품 정보 업데이트
    """
    try:
        # 업데이트할 필드 확인
        update_expression = "set "
        expression_attribute_values = {}
        
        for key, value in product_data.items():
            if key != 'id':  # ID는 업데이트 불가
                update_expression += f"{key} = :{key}, "
                expression_attribute_values[f":{key}"] = value
        
        # 마지막 쉼표 제거
        update_expression = update_expression[:-2]
        
        # 업데이트 시간 추가
        update_expression += ", updated_at = :updated_at"
        expression_attribute_values[":updated_at"] = datetime.now().isoformat()
        
        # DynamoDB 업데이트
        response = table.update_item(
            Key={'id': product_id},
            UpdateExpression=update_expression,
            ExpressionAttributeValues=expression_attribute_values,
            ReturnValues="ALL_NEW"
        )
        
        return {
            'statusCode': 200,
            'body': json.dumps(response['Attributes'])
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def delete_product(product_id):
    """
    제품 삭제
    """
    try:
        table.delete_item(Key={'id': product_id})
        
        return {
            'statusCode': 204,
            'body': ''
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
      

# virtual-service.yaml
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: product-service
spec:
  hosts:
  - product-service
  http:
  - route:
    - destination:
        host: product-service
        subset: v1
      weight: 90
    - destination:
        host: product-service
        subset: v2
      weight: 10
    retries:
      attempts: 3
      perTryTimeout: 2s
    timeout: 5s
---
# destination-rule.yaml
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: product-service
spec:
  host: product-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 1
        maxRequestsPerConnection: 10
    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
  subsets:
  - name: v1
    labels:
      version: v1
  - name: v2
    labels:
      version: v2
      

# argocd-application.yaml
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: shopping-microservices
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/your-org/shopping-microservices.git
    targetRevision: HEAD
    path: kubernetes
  destination:
    server: https://kubernetes.default.svc
    namespace: shopping
  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
    - CreateNamespace=true
    retry:
      limit: 5
      backoff:
        duration: 5s
        factor: 2
        maxDuration: 3m
      

# anomaly_detection.py
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from prometheus_client import Gauge, start_http_server
import time
import requests

# 메트릭 정의
ANOMALY_SCORE = Gauge('service_anomaly_score', 'Anomaly score for service metrics', ['service'])

# Prometheus 서버에서 메트릭 가져오기
def fetch_metrics(prometheus_url, query, start_time, end_time, step='1m'):
    response = requests.get(
        f"{prometheus_url}/api/v1/query_range",
        params={
            'query': query,
            'start': start_time,
            'end': end_time,
            'step': step
        }
    )
    return response.json()

# 이상 감지 모델 학습 및 예측
def detect_anomalies(service_name, metrics_data):
    # 데이터 준비
    df = pd.DataFrame(metrics_data)
    
    # 결측값 처리
    df = df.fillna(method='ffill')
    
    # 이상 감지 모델 학습
    model = IsolationForest(contamination=0.05, random_state=42)
    model.fit(df)
    
    # 이상 점수 계산 (-1에 가까울수록 이상)
    scores = model.decision_function(df)
    
    # 가장 최근 데이터의 이상 점수
    latest_score = scores[-1]
    
    # Prometheus 메트릭 업데이트
    ANOMALY_SCORE.labels(service=service_name).set(latest_score)
    
    return latest_score, latest_score < -0.5  # -0.5보다 작으면 이상으로 판단

# 메인 루프
def main():
    # Prometheus 메트릭 서버 시작
    start_http_server(8000)
    
    prometheus_url = "http://prometheus:9090"
    
    while True:
        try:
            # 현재 시간 및 1시간 전 시간
            end_time = int(time.time())
            start_time = end_time - 3600
            
            # 서비스별 메트릭 수집 및 이상 감지
            services = ["user-service", "product-service", "order-service", "payment-service"]
            
            for service in services:
                # CPU 사용량 메트릭
                cpu_query = f'container_cpu_usage_seconds_total{{pod=~"{service}-.*"}}'
                cpu_data = fetch_metrics(prometheus_url, cpu_query, start_time, end_time)
                
                # 메모리 사용량 메트릭
                memory_query =  <section class="container" style="max-width: 100%;">
  <section id="future-trends" style="margin-top: -20px;">
    <div class="code-container" style="margin-bottom: 30px;">
      <pre><code>
# anomaly_detection.py (계속)
                memory_query = f'container_memory_usage_bytes{{pod=~"{service}-.*"}}'
                memory_data = fetch_metrics(prometheus_url, memory_query, start_time, end_time)
                
                # 요청 지연 시간 메트릭
                latency_query = f'http_request_duration_seconds_sum{{service="{service}"}}'
                latency_data = fetch_metrics(prometheus_url, latency_query, start_time, end_time)
                
                # 오류율 메트릭
                error_query = f'http_requests_total{{service="{service}", status=~"5.."}}'
                error_data = fetch_metrics(prometheus_url, error_query, start_time, end_time)
                
                # 메트릭 데이터 결합
                metrics = []
                for result in cpu_data.get('data', {}).get('result', []):
                    metrics.append([float(v[1]) for v in result.get('values', [])])
                
                for result in memory_data.get('data', {}).get('result', []):
                    metrics.append([float(v[1]) for v in result.get('values', [])])
                
                for result in latency_data.get('data', {}).get('result', []):
                    metrics.append([float(v[1]) for v in result.get('values', [])])
                
                for result in error_data.get('data', {}).get('result', []):
                    metrics.append([float(v[1]) for v in result.get('values', [])])
                
                # 이상 감지 수행
                if metrics:
                    metrics_array = np.array(metrics).T  # 행렬 변환
                    score, is_anomaly = detect_anomalies(service, metrics_array)
                    
                    print(f"Service: {service}, Anomaly Score: {score:.4f}, Is Anomaly: {is_anomaly}")
                    
                    # 이상이 감지되면 알림 발송 (실제로는 Slack, Email 등으로 알림)
                    if is_anomaly:
                        print(f"ALERT: Anomaly detected in {service}!")
                        # send_alert(service, score, metrics_array[-1])
            
            # 5분마다 실행
            time.sleep(300)
        
        except Exception as e:
            print(f"Error in anomaly detection: {str(e)}")
            time.sleep(60)  # 오류 발생 시 1분 후 재시도

if __name__ == "__main__":
    main()