# 가상환경 생성 (권장)
python -m venv venv

# 가상환경 활성화
# Windows
venv\Scripts\activate
# Mac/Linux
source venv/bin/activate

# FastAPI 설치
pip install fastapi

# ASGI 서버 설치 (Uvicorn)
pip install "uvicorn[standard]"

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

uvicorn main:app --reload

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

app = FastAPI()

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None
    age: int = Field(..., ge=0, le=150)
    created_at: datetime = Field(default_factory=datetime.now)
    
    class Config:
        schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "john@example.com",
                "full_name": "John Doe",
                "age": 30
            }
        }

@app.post("/users/")
def create_user(user: User):
    return {
        "message": "User created successfully",
        "user": user
    }

from typing import List
from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

class Skill(BaseModel):
    name: str
    level: int = Field(..., ge=1, le=10)

class UserProfile(BaseModel):
    user: User
    address: Address
    skills: List[Skill]
    bio: Optional[str] = None

@app.post("/profiles/")
def create_profile(profile: UserProfile):
    return profile

{
  "user": {
    "username": "developer123",
    "email": "dev@example.com",
    "age": 28
  },
  "address": {
    "street": "123 Tech Street",
    "city": "Seoul",
    "country": "South Korea",
    "postal_code": "12345"
  },
  "skills": [
    {"name": "Python", "level": 9},
    {"name": "FastAPI", "level": 8}
  ],
  "bio": "Passionate developer"
}

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(..., title="아이템 ID", ge=1, le=1000)
):
    return {"item_id": item_id}

@app.get("/users/{user_id}/posts/{post_id}")
def read_user_post(
    user_id: int = Path(..., description="사용자 ID"),
    post_id: int = Path(..., description="게시글 ID")
):
    return {
        "user_id": user_id,
        "post_id": post_id,
        "message": f"User {user_id}의 Post {post_id}"
    }

from fastapi import Query
from typing import Optional, List

@app.get("/items/")
def list_items(
    skip: int = Query(0, ge=0, description="건너뛸 항목 수"),
    limit: int = Query(10, ge=1, le=100, description="가져올 항목 수"),
    search: Optional[str] = Query(None, min_length=3, max_length=50),
    tags: List[str] = Query([], description="태그 필터")
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search,
        "tags": tags
    }

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

@app.put("/items/{item_id}")
def update_item(
    item_id: int,
    item: Item,
    q: Optional[str] = None
):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

import time
import asyncio
from fastapi import FastAPI

app = FastAPI()

# 동기 함수
@app.get("/sync")
def sync_endpoint():
    time.sleep(2)  # 2초 대기 (블로킹)
    return {"message": "동기 처리 완료"}

# 비동기 함수
@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(2)  # 2초 대기 (논블로킹)
    return {"message": "비동기 처리 완료"}

from fastapi import FastAPI
from typing import List
import asyncio

app = FastAPI()

# 가상의 비동기 데이터베이스 함수
async def fetch_user_from_db(user_id: int):
    await asyncio.sleep(0.1)  # DB 쿼리 시뮬레이션
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_posts_from_db(user_id: int):
    await asyncio.sleep(0.1)  # DB 쿼리 시뮬레이션
    return [
        {"id": 1, "title": "Post 1"},
        {"id": 2, "title": "Post 2"}
    ]

@app.get("/users/{user_id}/complete")
async def get_user_complete(user_id: int):
    # 병렬로 실행!
    user, posts = await asyncio.gather(
        fetch_user_from_db(user_id),
        fetch_posts_from_db(user_id)
    )
    
    return {
        "user": user,
        "posts": posts
    }

pip install sqlalchemy databases asyncpg

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLite 사용 (개발용)
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# PostgreSQL 사용 (프로덕션)
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite용
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

from sqlalchemy import Boolean, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    author = relationship("User", back_populates="posts")

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import List, Optional

# 기본 스키마
class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    author_id: int
    created_at: datetime
    
    class Config:
        orm_mode = True  # SQLAlchemy 모델과 호환

class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    posts: List[Post] = []
    
    class Config:
        orm_mode = True

from sqlalchemy.orm import Session
from typing import List, Optional
import models, schemas
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def get_user(db: Session, user_id: int) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[models.User]:
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate) -> models.User:
    hashed_password = get_password_hash(user.password)
    db_user = models.User(
        email=user.email,
        username=user.username,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def create_post(db: Session, post: schemas.PostCreate, user_id: int) -> models.Post:
    db_post = models.Post(**post.dict(), author_id=user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def get_posts(db: Session, skip: int = 0, limit: int = 100) -> List[models.Post]:
    return db.query(models.Post).offset(skip).limit(limit).all()

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import crud, models, schemas
from database import SessionLocal, engine, get_db

# 테이블 생성
models.Base.metadata.create_all(bind=engine)

app = FastAPI(title="블로그 API", version="1.0.0")

# 사용자 관련 엔드포인트
@app.post("/users/", response_model=schemas.User, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="이미 등록된 이메일입니다")
    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    return db_user

# 게시글 관련 엔드포인트
@app.post("/users/{user_id}/posts/", response_model=schemas.Post, status_code=201)
def create_post_for_user(
    user_id: int,
    post: schemas.PostCreate,
    db: Session = Depends(get_db)
):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    return crud.create_post(db=db, post=post, user_id=user_id)

@app.get("/posts/", response_model=List[schemas.Post])
def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    posts = crud.get_posts(db, skip=skip, limit=limit)
    return posts

pip install python-jose[cryptography] passlib[bcrypt] python-multipart

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

# 설정값 (실제로는 환경변수로 관리!)
SECRET_KEY = "your-secret-key-keep-it-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[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

def decode_access_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
import crud, models, schemas
from database import get_db
from security import (
    verify_password,
    create_access_token,
    ACCESS_TOKEN_EXPIRE_MINUTES
)

app = FastAPI()

@app.post("/token")
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    # 사용자 확인
    user = crud.get_user_by_email(db, email=form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="이메일 또는 비밀번호가 올바르지 않습니다",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 액세스 토큰 생성
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.email},
        expires_delta=access_token_expires
    )
    
    return {
        "access_token": access_token,
        "token_type": "bearer"
    }

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from security import oauth2_scheme, decode_access_token
from database import get_db
import crud

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="인증 정보를 확인할 수 없습니다",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    
    email: str = payload.get("sub")
    if email is None:
        raise credentials_exception
    
    user = crud.get_user_by_email(db, email=email)
    if user is None:
        raise credentials_exception
    
    return user

async def get_current_active_user(
    current_user: models.User = Depends(get_current_user)
):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="비활성화된 사용자입니다")
    return current_user

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

@app.post("/posts/", response_model=schemas.Post)
async def create_post(
    post: schemas.PostCreate,
    current_user: models.User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    return crud.create_post(db=db, post=post, user_id=current_user.id)

@app.delete("/posts/{post_id}")
async def delete_post(
    post_id: int,
    current_user: models.User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    post = db.query(models.Post).filter(models.Post.id == post_id).first()
    if not post:
        raise HTTPException(status_code=404, detail="게시글을 찾을 수 없습니다")
    
    if post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="권한이 없습니다")
    
    db.delete(post)
    db.commit()
    return {"message": "게시글이 삭제되었습니다"}

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id}를 찾을 수 없습니다",
            headers={"X-Error": "Item not found"}
        )
    return items_db[item_id]

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# 커스텀 예외 클래스
class ItemNotFoundException(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

class InsufficientPermissionException(Exception):
    def __init__(self, message: str = "권한이 부족합니다"):
        self.message = message

# 예외 핸들러 등록
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={
            "error": "ItemNotFound",
            "message": f"Item {exc.item_id}를 찾을 수 없습니다",
            "item_id": exc.item_id
        }
    )

@app.exception_handler(InsufficientPermissionException)
async def permission_handler(request: Request, exc: InsufficientPermissionException):
    return JSONResponse(
        status_code=status.HTTP_403_FORBIDDEN,
        content={
            "error": "InsufficientPermission",
            "message": exc.message
        }
    )

# 검증 에러 커스터마이징
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "ValidationError",
            "message": "입력 데이터가 올바르지 않습니다",
            "details": exc.errors()
        }
    )

# 사용 예시
@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id not in items_db:
        raise ItemNotFoundException(item_id)
    return items_db[item_id]

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger(__name__)

app = FastAPI()

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unexpected error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "InternalServerError",
            "message": "서버 내부 오류가 발생했습니다. 잠시 후 다시 시도해주세요."
        }
    )

from sqlalchemy.orm import Session, joinedload

# ❌ 나쁜 예: N+1 문제
@app.get("/users-with-posts-bad")
def get_users_bad(db: Session = Depends(get_db)):
    users = db.query(User).all()
    result = []
    for user in users:  # N번의 추가 쿼리 발생!
        result.append({
            "user": user,
            "posts": user.posts  # 각 사용자마다 쿼리 실행
        })
    return result

# ✅ 좋은 예: Eager Loading
@app.get("/users-with-posts-good")
def get_users_good(db: Session = Depends(get_db)):
    users = db.query(User).options(
        joinedload(User.posts)  # 한 번의 JOIN 쿼리로 해결!
    ).all()
    return users

from functools import lru_cache
from fastapi import FastAPI
import redis
import json

app = FastAPI()

# 메모리 캐싱 (간단한 경우)
@lru_cache(maxsize=128)
def get_expensive_data(key: str):
    # 비용이 큰 연산
    return expensive_computation(key)

# Redis 캐싱 (분산 환경)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.get("/products/{product_id}")
async def get_product(product_id: int, db: Session = Depends(get_db)):
    # 캐시 확인
    cache_key = f"product:{product_id}"
    cached = redis_client.get(cache_key)
    
    if cached:
        return json.loads(cached)
    
    # DB에서 조회
    product = db.query(Product).filter(Product.id == product_id).first()
    if not product:
        raise HTTPException(status_code=404)
    
    # 캐시 저장 (1시간)
    redis_client.setex(
        cache_key,
        3600,
        json.dumps(product.dict())
    )
    
    return product

from typing import Generic, TypeVar, List
from pydantic import BaseModel
from math import ceil

T = TypeVar('T')

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    page_size: int
    total_pages: int

@app.get("/items/", response_model=PaginatedResponse[schemas.Item])
def list_items(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    db: Session = Depends(get_db)
):
    # 전체 개수 조회
    total = db.query(Item).count()
    
    # 페이지네이션 적용
    items = db.query(Item)\
        .offset((page - 1) * page_size)\
        .limit(page_size)\
        .all()
    
    return PaginatedResponse(
        items=items,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=ceil(total / page_size)
    )

from fastapi import BackgroundTasks
import time

def send_email(email: str, message: str):
    time.sleep(3)  # 이메일 전송 시뮬레이션
    print(f"Email sent to {email}: {message}")

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    # 백그라운드에서 실행
    background_tasks.add_task(send_email, email, "Welcome!")
    
    # 즉시 응답 반환
    return {"message": "알림이 전송됩니다"}

import time
from fastapi import Request

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    print(f"{request.method} {request.url.path}: {process_time:.4f}s")
    return response

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

pip install pytest httpx

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app, get_db
from database import Base

# 테스트용 데이터베이스
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 테이블 생성
Base.metadata.create_all(bind=engine)

# 의존성 오버라이드
def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

# 테스트 클라이언트
client = TestClient(app)

# 테스트 케이스
def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

def test_create_user():
    response = client.post(
        "/users/",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "testpass123"
        }
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "id" in data

def test_create_user_duplicate_email():
    # 첫 번째 사용자 생성
    client.post(
        "/users/",
        json={
            "email": "duplicate@example.com",
            "username": "user1",
            "password": "pass123"
        }
    )
    
    # 같은 이메일로 다시 생성 시도
    response = client.post(
        "/users/",
        json={
            "email": "duplicate@example.com",
            "username": "user2",
            "password": "pass456"
        }
    )
    assert response.status_code == 400
    assert "이미 등록된" in response.json()["detail"]

def test_read_user():
    # 사용자 생성
    create_response = client.post(
        "/users/",
        json={
            "email": "read@example.com",
            "username": "readuser",
            "password": "pass123"
        }
    )
    user_id = create_response.json()["id"]
    
    # 사용자 조회
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == user_id
    assert data["email"] == "read@example.com"

def test_read_nonexistent_user():
    response = client.get("/users/99999")
    assert response.status_code == 404

def test_login():
    # 사용자 생성
    client.post(
        "/users/",
        json={
            "email": "login@example.com",
            "username": "loginuser",
            "password": "password123"
        }
    )
    
    # 로그인
    response = client.post(
        "/token",
        data={
            "username": "login@example.com",
            "password": "password123"
        }
    )
    assert response.status_code == 200
    assert "access_token" in response.json()
    return response.json()["access_token"]

def test_protected_endpoint():
    token = test_login()
    
    # 인증된 요청
    response = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    assert response.json()["email"] == "login@example.com"

def test_protected_endpoint_without_token():
    response = client.get("/users/me")
    assert response.status_code == 401

import pytest

@pytest.fixture
def test_user():
    response = client.post(
        "/users/",
        json={
            "email": "fixture@example.com",
            "username": "fixtureuser",
            "password": "pass123"
        }
    )
    return response.json()

@pytest.fixture
def auth_token(test_user):
    response = client.post(
        "/token",
        data={
            "username": test_user["email"],
            "password": "pass123"
        }
    )
    return response.json()["access_token"]

def test_with_fixtures(test_user, auth_token):
    response = client.get(
        f"/users/{test_user['id']}",
        headers={"Authorization": f"Bearer {auth_token}"}
    )
    assert response.status_code == 200

# 모든 테스트 실행
pytest

# 특정 파일만 실행
pytest test_main.py

# 상세 출력
pytest -v

# 커버리지 확인
pytest --cov=main

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(
    title="블로그 API",
    description="""
    ## 소개
    이 API는 블로그 플랫폼을 위한 RESTful API입니다.
    
    ## 주요 기능
    * **사용자 관리**: 회원가입, 로그인, 프로필 관리
    * **게시글**: CRUD 작업, 검색, 필터링
    * **댓글**: 게시글에 댓글 작성 및 관리
    
    ## 인증
    JWT Bearer 토큰을 사용합니다.
    """,
    version="1.0.0",
    terms_of_service="https://example.com/terms/",
    contact={
        "name": "API Support",
        "url": "https://example.com/support",
        "email": "support@example.com"
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT"
    }
)

class Item(BaseModel):
    name: str = Field(..., description="아이템 이름", example="노트북")
    price: float = Field(..., gt=0, description="가격 (원)", example=1500000)
    description: str = Field(None, description="상세 설명", example="고성능 노트북")

@app.post(
    "/items/",
    response_model=Item,
    status_code=201,
    summary="아이템 생성",
    description="새로운 아이템을 생성합니다.",
    response_description="생성된 아이템 정보",
    tags=["items"]
)
def create_item(item: Item):
    """
    아이템을 생성합니다:
    
    - **name**: 필수, 아이템 이름
    - **price**: 필수, 0보다 큰 가격
    - **description**: 선택, 상세 설명
    """
    return item

tags_metadata = [
    {
        "name": "users",
        "description": "사용자 관련 작업",
    },
    {
        "name": "items",
        "description": "아이템 관리",
        "externalDocs": {
            "description": "외부 문서",
            "url": "https://example.com/docs/items",
        },
    },
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
def get_users():
    return []

@app.get("/items/", tags=["items"])
def get_items():
    return []

from fastapi import FastAPI, APIRouter

app = FastAPI()

# v1 라우터
v1_router = APIRouter(prefix="/api/v1", tags=["v1"])

@v1_router.get("/users/")
def get_users_v1():
    return {"version": "1.0", "users": []}

# v2 라우터 (개선된 버전)
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])

@v2_router.get("/users/")
def get_users_v2(
    page: int = 1,
    page_size: int = 20
):
    return {
        "version": "2.0",
        "users": [],
        "pagination": {
            "page": page,
            "page_size": page_size
        }
    }

# 라우터 등록
app.include_router(v1_router)
app.include_router(v2_router)

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",  # React 개발 서버
        "http://localhost:8080",  # Vue 개발 서버
        "https://yourdomain.com"  # 프로덕션 도메인
    ],
    allow_credentials=True,
    allow_methods=["*"],  # 모든 HTTP 메서드 허용
    allow_headers=["*"],  # 모든 헤더 허용
)

# 개발 환경에서는 모든 origin 허용 (주의!)
# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=["*"],
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )

from fastapi import FastAPI, Request
import time
import logging

logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    # 요청 시작 시간
    start_time = time.time()
    
    # 요청 정보 로깅
    logger.info(f"Request: {request.method} {request.url}")
    
    # 다음 미들웨어 또는 엔드포인트 실행
    response = await call_next(request)
    
    # 처리 시간 계산
    process_time = time.time() - start_time
    
    # 응답 헤더에 처리 시간 추가
    response.headers["X-Process-Time"] = str(process_time)
    
    # 응답 정보 로깅
    logger.info(
        f"Response: {response.status_code} "
        f"(took {process_time:.4f}s)"
    )
    
    return response

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    
    # 보안 헤더 추가
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    
    return response

from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
from datetime import datetime, timedelta
import asyncio

app = FastAPI()

# 간단한 인메모리 rate limiter
class RateLimiter:
    def __init__(self, requests: int, window: int):
        self.requests = requests  # 허용 요청 수
        self.window = window  # 시간 윈도우 (초)
        self.clients = defaultdict(list)
    
    async def check_rate_limit(self, client_id: str):
        now = datetime.now()
        # 오래된 요청 제거
        self.clients[client_id] = [
            req_time for req_time in self.clients[client_id]
            if now - req_time < timedelta(seconds=self.window)
        ]
        
        # 요청 수 확인
        if len(self.clients[client_id]) >= self.requests:
            return False
        
        # 새 요청 기록
        self.clients[client_id].append(now)
        return True

rate_limiter = RateLimiter(requests=10, window=60)  # 1분에 10개 요청

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    
    if not await rate_limiter.check_rate_limit(client_ip):
        raise HTTPException(
            status_code=429,
            detail="요청 한도를 초과했습니다. 잠시 후 다시 시도해주세요."
        )
    
    response = await call_next(request)
    return response

pip install python-dotenv

# .env
DATABASE_URL=postgresql://user:password@localhost/dbname
SECRET_KEY=your-super-secret-key-here
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
REDIS_URL=redis://localhost:6379
ENVIRONMENT=production

from pydantic import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    redis_url: str
    environment: str = "development"
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings():
    return Settings()

# 사용
settings = get_settings()

# Dockerfile
FROM python:3.9-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"]

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/dbname
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    volumes:
      - .:/app
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
  
  db:
    image: postgres:13
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=dbname
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:6-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:

pip install gunicorn

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --access-logfile - \
  --error-logfile -

# /etc/nginx/sites-available/fastapi
server {
    listen 80;
    server_name api.yourdomain.com;
    
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
    
    # 정적 파일
    location /static {
        alias /var/www/static;
    }
}

import logging
from logging.handlers import RotatingFileHandler

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

# 파일 핸들러 (로그 로테이션)
handler = RotatingFileHandler(
    'app.log',
    maxBytes=10485760,  # 10MB
    backupCount=10
)
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# 사용
@app.get("/")
def root():
    logger.info("Root endpoint accessed")
    return {"message": "Hello"}

pip install sentry-sdk

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[FastApiIntegration()],
    traces_sample_rate=1.0,
    environment=settings.environment
)

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False, index=True)
    slug = Column(String(200), unique=True, index=True)
    content = Column(Text, nullable=False)
    excerpt = Column(String(500))
    published = Column(Boolean, default=False)
    views = Column(Integer, default=0)
    
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 관계
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")

from pydantic import BaseModel, Field, validator
from typing import List, Optional
from datetime import datetime
import re

class PostBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1)
    excerpt: Optional[str] = Field(None, max_length=500)
    published: bool = False
    tags: List[str] = []

class PostCreate(PostBase):
    @validator('title')
    def generate_slug(cls, v):
        # 슬러그 자동 생성 로직
        return re.sub(r'[^\w\s-]', '', v.lower()).replace(' ', '-')

class PostUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    content: Optional[str] = Field(None, min_length=1)
    excerpt: Optional[str] = Field(None, max_length=500)
    published: Optional[bool] = None
    tags: Optional[List[str]] = None

class PostInDB(PostBase):
    id: int
    slug: str
    views: int
    author_id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        orm_mode = True

class Post(PostInDB):
    author: "UserBasic"
    comments_count: int = 0
    
class PostDetail(Post):
    comments: List["Comment"] = []

from sqlalchemy.orm import Session
from sqlalchemy import or_
from typing import List, Optional
from app.models.post import Post
from app.schemas.post import PostCreate, PostUpdate

def get_post(db: Session, post_id: int) -> Optional[Post]:
    return db.query(Post).filter(Post.id == post_id).first()

def get_post_by_slug(db: Session, slug: str) -> Optional[Post]:
    return db.query(Post).filter(Post.slug == slug).first()

def get_posts(
    db: Session,
    skip: int = 0,
    limit: int = 20,
    published_only: bool = True,
    search: Optional[str] = None,
    author_id: Optional[int] = None
) -> List[Post]:
    query = db.query(Post)
    
    if published_only:
        query = query.filter(Post.published == True)
    
    if author_id:
        query = query.filter(Post.author_id == author_id)
    
    if search:
        query = query.filter(
            or_(
                Post.title.ilike(f"%{search}%"),
                Post.content.ilike(f"%{search}%")
            )
        )
    
    return query.order_by(Post.created_at.desc()).offset(skip).limit(limit).all()

def create_post(db: Session, post: PostCreate, author_id: int) -> Post:
    db_post = Post(
        **post.dict(exclude={'tags'}),
        author_id=author_id
    )
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def update_post(db: Session, post_id: int, post: PostUpdate) -> Optional[Post]:
    db_post = get_post(db, post_id)
    if not db_post:
        return None
    
    update_data = post.dict(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_post, field, value)
    
    db.commit()
    db.refresh(db_post)
    return db_post

def delete_post(db: Session, post_id: int) -> bool:
    db_post = get_post(db, post_id)
    if not db_post:
        return False
    
    db.delete(db_post)
    db.commit()
    return True

def increment_views(db: Session, post_id: int):
    db_post = get_post(db, post_id)
    if db_post:
        db_post.views += 1
        db.commit()

from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List
from app import crud, schemas
from app.api.deps import get_db, get_current_active_user
from app.models.user import User

router = APIRouter()

@router.get("/", response_model=List[schemas.Post])
def list_posts(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    search: str = Query(None),
    db: Session = Depends(get_db)
):
    """게시글 목록 조회"""
    posts = crud.get_posts(
        db,
        skip=skip,
        limit=limit,
        search=search,
        published_only=True
    )
    return posts

@router.get("/{post_id}", response_model=schemas.PostDetail)
def get_post(
    post_id: int,
    db: Session = Depends(get_db)
):
    """게시글 상세 조회"""
    post = crud.get_post(db, post_id)
    if not post or not post.published:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="게시글을 찾을 수 없습니다"
        )
    
    # 조회수 증가
    crud.increment_views(db, post_id)
    
    return post

@router.post("/", response_model=schemas.Post, status_code=status.HTTP_201_CREATED)
def create_post(
    post: schemas.PostCreate,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    """게시글 작성"""
    return crud.create_post(db, post, current_user.id)

@router.put("/{post_id}", response_model=schemas.Post)
def update_post(
    post_id: int,
    post: schemas.PostUpdate,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    """게시글 수정"""
    db_post = crud.get_post(db, post_id)
    if not db_post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="게시글을 찾을 수 없습니다"
        )
    
    if db_post.author_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="권한이 없습니다"
        )
    
    return crud.update_post(db, post_id, post)

@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(
    post_id: int,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    """게시글 삭제"""
    db_post = crud.get_post(db, post_id)
    if not db_post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="게시글을 찾을 수 없습니다"
        )
    
    if db_post.author_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="권한이 없습니다"
        )
    
    crud.delete_post(db, post_id)
    return None

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left")

from fastapi import FastAPI, File, UploadFile, HTTPException
from typing import List
import shutil
from pathlib import Path

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # 파일 크기 제한 (10MB)
    max_size = 10 * 1024 * 1024
    contents = await file.read()
    
    if len(contents) > max_size:
        raise HTTPException(
            status_code=413,
            detail="파일 크기가 너무 큽니다 (최대 10MB)"
        )
    
    # 허용된 확장자 체크
    allowed_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.pdf'}
    file_ext = Path(file.filename).suffix.lower()
    
    if file_ext not in allowed_extensions:
        raise HTTPException(
            status_code=400,
            detail=f"허용되지 않는 파일 형식입니다. 허용: {allowed_extensions}"
        )
    
    # 파일 저장
    file_path = UPLOAD_DIR / file.filename
    with open(file_path, "wb") as buffer:
        buffer.write(contents)
    
    return {
        "filename": file.filename,
        "size": len(contents),
        "content_type": file.content_type
    }

@app.post("/upload-multiple/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        file_path = UPLOAD_DIR / file.filename
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        results.append({"filename": file.filename})
    
    return {"uploaded": results}

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    print("애플리케이션 시작!")
    # 데이터베이스 연결 풀 생성
    # 캐시 초기화
    # 백그라운드 작업 시작
    asyncio.create_task(background_task())

@app.on_event("shutdown")
async def shutdown_event():
    print("애플리케이션 종료!")
    # 데이터베이스 연결 정리
    # 캐시 저장
    # 리소스 해제

async def background_task():
    while True:
        print("백그라운드 작업 실행 중...")
        await asyncio.sleep(60)

pip install strawberry-graphql

import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class User:
    id: int
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: int) -> User:
        return User(id=id, name="John", email="john@example.com")
    
    @strawberry.field
    def users(self) -> list[User]:
        return [
            User(id=1, name="John", email="john@example.com"),
            User(id=2, name="Jane", email="jane@example.com")
        ]

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")