Python ๋น๋๊ธฐ(ASGI) REST API, Uvicorn ์คํ, Pydantic v2 ๋ชจ๋ธ ๊ฒ์ฆ, ๊ฒฝ๋ก/์ฟผ๋ฆฌ ํ๋ผ๋ฏธํฐ, ์์กด์ฑ ์ฃผ์
(Depends), SQLAlchemy 2.0 ๋น๋๊ธฐ DB ์ฐ๋, JWT ์ธ์ฆ, ๋ผ์ฐํฐ ๋ชจ๋ํ ๋ฐ ํ๋ก๋์
๋ฐฐํฌ ๋ ํผ๋ฐ์ค
## 1. ํ๋ก์ ํธ ์์ & ๊ตฌ์กฐ (Project Setup & Architecture)
```bash
# โก [1] uv ํจํค์ง ๋งค๋์ ๋ก ํ๋ก์ ํธ ์ด๊ธฐํ ๋ฐ FastAPI ์ค์น (๊ฐ์ฅ ๊ถ์ฅ)
uv init my-fastapi-app
cd my-fastapi-app
uv add "fastapi[standard]" # FastAPI, Uvicorn, Pydantic, Email-validator ๋ฑ ํ์ค ๋ฒ๋ค ์ผ๊ด ์ค์น
uv add "sqlalchemy>=2.0" asyncpg aiosqlite pydantic-settings pyjwt passlib[bcrypt]
# ๐ฆ [2] pip ๋๋ pipenv ํ๊ฒฝ ์ค์น
pip install "fastapi[standard]" uvicorn
```
```text
my-fastapi-app/
โโโ app/
โ โโโ main.py # FastAPI ์ธ์คํด์ค ์ด๊ธฐํ & ๋ฏธ๋ค์จ์ด & lifespan ์ง์
์
โ โโโ core/
โ โ โโโ config.py # Pydantic Settings ํ๊ฒฝ ๋ณ์ (.env) ๋ก๋
โ โ โโโ database.py # SQLAlchemy 2.0 ๋น๋๊ธฐ ์์ง & ์ธ์
ํฉํ ๋ฆฌ
โ โ โโโ security.py # JWT ํ ํฐ ์ธ์ฝ๋ฉ/๋์ฝ๋ฉ & ํจ์ค์๋ ํด์ฑ
โ โโโ api/
โ โ โโโ v1/
โ โ โ โโโ api.py # v1 ๋ผ์ฐํฐ ํตํฉ ๋ฌถ์ (APIRouter)
โ โ โ โโโ endpoints/
โ โ โ โโโ auth.py # ๋ก๊ทธ์ธ, ํ ํฐ ๋ฐ๊ธ ์๋ํฌ์ธํธ
โ โ โ โโโ users.py # ์ฌ์ฉ์ CRUD API
โ โ โ โโโ items.py # ๋น์ฆ๋์ค ๋ฆฌ์์ค API
โ โ โโโ deps.py # ์ ์ญ ์์กด์ฑ ์ฃผ์
(get_db, get_current_user)
โ โโโ models/ # SQLAlchemy ORM ํ
์ด๋ธ ๋งคํ ํด๋์ค
โ โ โโโ user.py
โ โ โโโ item.py
โ โโโ schemas/ # Pydantic v2 Request/Response DTO ์คํค๋ง
โ โ โโโ user.py
โ โ โโโ item.py
โ โโโ services/ # ๋น์ฆ๋์ค ๋น๋๊ธฐ ์ฒ๋ฆฌ ๋ก์ง
โโโ tests/ # pytest ๋ฐ TestClient ํ
์คํธ ์ค์ํธ
โโโ .env # ๊ฐ๋ฐ ํ๊ฒฝ ๋น๋ฐํค ๋ฐ DB ์ ์ URI
โโโ Dockerfile # ํ๋ก๋์
์ปจํ
์ด๋ ๋น๋ ํ์ผ
โโโ pyproject.toml # ์์กด์ฑ ๋ฐ ํ๋ก์ ํธ ๋ฉํ๋ฐ์ดํฐ
```
---
## 2. ๊ธฐ๋ณธ ์ฑ & ์คํ (Basic App & Uvicorn CLI)
```python
# app/main.py
from fastapi import FastAPI, status
# FastAPI ์ธ์คํด์ค ์์ฑ ๋ฐ ๋ฉํ๋ฐ์ดํฐ ์ ์ (Swagger UI ์๋ ๋ฐ์)
app = FastAPI(
title="LuckyTech API Service",
description="FastAPI 0.115+ ๊ธฐ๋ฐ ๊ณ ์ฑ๋ฅ ๋น๋๊ธฐ REST API ๋ฐฑ์๋",
version="1.0.0",
docs_url="/docs", # Swagger UI ์๋ํฌ์ธํธ (์ด์ ํ๊ฒฝ์์๋ None์ผ๋ก ๋นํ์ฑํ ๊ฐ๋ฅ)
redoc_url="/redoc", # ReDoc ์๋ํฌ์ธํธ
openapi_url="/api/v1/openapi.json"
)
# ํฌ์ค์ฒดํฌ & ๊ธฐ๋ณธ GET ์๋ํฌ์ธํธ
@app.get("/", tags=["Health"], summary="์๋น์ค ํฌ์ค์ฒดํฌ")
async def health_check():
return {"status": "healthy", "service": "LuckyTech API", "version": "1.0.0"}
# HTTP ๋ฉ์๋๋ณ ๊ธฐ๋ณธ ๋ผ์ฐํ
์์
@app.get("/items", tags=["Items"])
async def list_items():
return [{"id": 1, "name": "Keyboard"}]
@app.post("/items", status_code=status.HTTP_201_CREATED, tags=["Items"])
async def create_item(payload: dict):
return {"message": "Created", "data": payload}
```
```bash
# ๐ Uvicorn ๊ฐ๋ฐ ์๋ฒ ์คํ ๋ช
๋ น์ด
# 1) FastAPI CLI (FastAPI 0.111+ ๋ด์ฅ ํ์ค ๋ช
๋ น์ด)
fastapi dev app/main.py # ๊ฐ๋ฐ ๋ชจ๋ (์๋ ๋ฆฌ๋ก๋, ๋ก์ปฌ ์ฃผ์ ๋ฐ์ธ๋ฉ)
fastapi run app/main.py --port 8000 # ํ๋ก๋์
๋ชจ๋
# 2) Uvicorn ์ง์ ์คํ
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
# ๐ ๋ํํ ๋ฌธ์ ๋ธ๋ผ์ฐ์ ์ ์
# http://localhost:8000/docs (Swagger UI - ํ
์คํธ ์คํ ๊ฐ๋ฅ)
# http://localhost:8000/redoc (ReDoc - ๋ช
์ธ์ ๋ทฐ์ด)
```
---
## 3. ์์ฒญ ํ๋ผ๋ฏธํฐ (Path, Query & Header)
```python
from typing import Annotated
from fastapi import APIRouter, Path, Query, Header, Cookie, HTTPException, status
router = APIRouter(prefix="/products", tags=["Products"])
# [1] ๊ฒฝ๋ก ํ๋ผ๋ฏธํฐ (Path Parameter) + ์ ํจ์ฑ ๊ฒ์ฆ
@router.get("/{product_id}")
async def get_product(
product_id: Annotated[int, Path(title="์ํ ๊ณ ์ ID", ge=1, le=999999, description="1 ์ด์์ ์ ์๊ฐ")]
):
return {"product_id": product_id, "name": f"Product-{product_id}"}
# [2] ์ฟผ๋ฆฌ ํ๋ผ๋ฏธํฐ (Query Parameter) + ๊ธฐ๋ณธ๊ฐ + ์ ํ์ ํํฐ
@router.get("/")
async def search_products(
keyword: Annotated[str | None, Query(min_length=2, max_length=50, description="๊ฒ์ ํค์๋")] = None,
page: Annotated[int, Query(ge=1, description="ํ์ด์ง ๋ฒํธ")] = 1,
size: Annotated[int, Query(ge=1, le=100, description="ํ์ด์ง ๋น ํญ๋ชฉ ์")] = 20,
tags: Annotated[list[str], Query(description="๋ค์ค ํ๊ทธ ํํฐ (?tags=tech&tags=python)")] = []
):
return {
"page": page,
"size": size,
"keyword": keyword,
"tags": tags,
"items": []
}
# [3] ์์ฒญ ํค๋ ๋ฐ ์ฟ ํค ์์
@router.get("/user-agent/check")
async def inspect_request(
user_agent: Annotated[str | None, Header(description="ํด๋ผ์ด์ธํธ User-Agent")] = None,
x_custom_token: Annotated[str | None, Header(alias="X-Custom-Token")] = None,
session_id: Annotated[str | None, Cookie()] = None
):
return {
"user_agent": user_agent,
"token": x_custom_token,
"has_session": session_id is not None
}
```
---
## 4. Pydantic v2 ์์ฒญ & ์๋ต ๋ชจ๋ธ (Request Body & Validation)
```python
from datetime import datetime
from pydantic import BaseModel, Field, EmailStr, field_validator, model_validator, ConfigDict
# [1] ๊ธฐ๋ณธ ์์ฒญ DTO ์คํค๋ง
class UserCreateRequest(BaseModel):
# Field: ๊ธฐ๋ณธ๊ฐ, ์ ์ฝ์กฐ๊ฑด, ๋ฉํ๋ฐ์ดํฐ ์ค์
username: str = Field(..., min_length=3, max_length=30, pattern=r"^[a-zA-Z0-9_-]+$", description="์ํ๋ฒณ, ์ซ์, _, - ํ์ฉ")
email: EmailStr = Field(..., description="์ ํจํ ์ด๋ฉ์ผ ์ฃผ์")
password: str = Field(..., min_length=8, description="8์ ์ด์์ ๋น๋ฐ๋ฒํธ")
confirm_password: str = Field(..., min_length=8)
age: int | None = Field(default=None, ge=1, le=150)
is_active: bool = Field(default=True)
# ๊ฐ๋ณ ํ๋ ์ ํจ์ฑ ๊ฒ์ฆ (@field_validator)
@field_validator("username")
@classmethod
def validate_username(cls, v: str) -> str:
forbidden = ["admin", "root", "system"]
if v.lower() in forbidden:
raise ValueError(f"'{v}'๋ ์์ฝ์ด๋ก ์ฌ์ฉํ ์ ์์ต๋๋ค.")
return v.lower()
# ๋ชจ๋ธ ์ ์ฒด ๋จ์ ๊ฒ์ฆ (๋น๋ฐ๋ฒํธ ์ผ์น ํ์ธ ๋ฑ)
@model_validator(mode="after")
def verify_password_match(self):
if self.password != self.confirm_password:
raise ValueError("๋น๋ฐ๋ฒํธ์ ๋น๋ฐ๋ฒํธ ํ์ธ์ด ์ผ์นํ์ง ์์ต๋๋ค.")
return self
# [2] ์๋ต DTO ์คํค๋ง (๋ฏผ๊ฐํ ๋น๋ฐ๋ฒํธ ์ ์ธ ๋ฐ ORM ํธํ ์ค์ )
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True) # SQLAlchemy ๋ชจ๋ธ ๊ฐ์ฒด๋ฅผ Pydantic ๋ชจ๋ธ๋ก ์๋ ๋ณํ (v1์ orm_mode=True)
id: int
username: str
email: EmailStr
age: int | None
is_active: bool
created_at: datetime
# [3] ์๋ํฌ์ธํธ ์ ์ฉ (์๋ ์ญ์ง๋ ฌํ & ์ ํจ์ฑ ๊ฒ์ฆ)
from fastapi import APIRouter
user_router = APIRouter(prefix="/users", tags=["Users"])
@user_router.post("/", response_model=UserResponse, status_code=201)
async def register_user(payload: UserCreateRequest):
# payload ์ ํจ์ฑ ๊ฒ์ฆ ์คํจ ์ FastAPI๊ฐ ์๋์ผ๋ก 422 Unprocessable Entity ๋ฐํ
return {
"id": 101,
"username": payload.username,
"email": payload.email,
"age": payload.age,
"is_active": payload.is_active,
"created_at": datetime.now()
}
```
---
## 5. ์๋ต ๋ชจ๋ธ & ์ํ ์ฝ๋ & ์์ธ ์ฒ๋ฆฌ (Response & Error Handling)
```python
from fastapi import FastAPI, HTTPException, status, Request
from fastapi.responses import JSONResponse, StreamingResponse, RedirectResponse
from pydantic import BaseModel
app = FastAPI()
# [1] ํ์ค HTTPException ๋ฐ์
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 404:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"error_code": "ITEM_NOT_FOUND", "message": f"ID {item_id} ์ํ์ด ์กด์ฌํ์ง ์์ต๋๋ค."},
headers={"X-Error-Reason": "Resource Missing"}
)
return {"id": item_id, "name": "Super Item"}
# [2] ์ปค์คํ
๋น์ฆ๋์ค ์์ธ & ์ ์ญ ํธ๋ค๋ฌ ๋ฑ๋ก
class BusinessException(Exception):
def __init__(self, code: str, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code
@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"code": exc.code,
"message": exc.message,
"path": str(request.url.path)
}
)
@app.post("/orders/pay")
async def process_payment(amount: int):
if amount <= 0:
raise BusinessException(code="INVALID_PAYMENT_AMOUNT", message="๊ฒฐ์ ๊ธ์ก์ 0์๋ณด๋ค ์ปค์ผ ํฉ๋๋ค.")
return {"success": True, "amount": amount}
# [3] ํ์ผ ์คํธ๋ฆฌ๋ฐ ์๋ต (๋์ฉ๋ ๋ก๊ทธ or CSV ๋ด๋ณด๋ด๊ธฐ)
async def fake_csv_generator():
yield "id,username,score\n"
for i in range(1, 100):
yield f"{i},user_{i},{i*10}\n"
@app.get("/export/csv")
async def export_csv():
return StreamingResponse(
fake_csv_generator(),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=report.csv"}
)
```
---
## 6. ์์กด์ฑ ์ฃผ์
์์คํ
(Dependency Injection: Depends)
```python
from typing import Annotated
from fastapi import Depends, Header, HTTPException, status
# [1] ๋จ์ ํจ์ ๊ธฐ๋ฐ ์์กด์ฑ (๊ณตํต ํค๋ ๊ฒ์ฆ)
async def verify_api_key(x_api_key: Annotated[str | None, Header()] = None):
if not x_api_key or x_api_key != "secret-lucky-token":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="์ ํจํ์ง ์๊ฑฐ๋ ๋๋ฝ๋ API ํค์
๋๋ค."
)
return x_api_key
# [2] ํด๋์ค ๊ธฐ๋ฐ ๊ณตํต ์ฟผ๋ฆฌ ํ์ด์ง๋ค์ด์
์์กด์ฑ
class PaginationParams:
def __init__(self, page: int = 1, size: int = 20):
self.page = max(1, page)
self.size = min(max(1, size), 100)
self.offset = (self.page - 1) * self.size
# [3] yield ๊ธฐ๋ฐ ๋ฆฌ์์ค ๋ผ์ดํ์ฌ์ดํด ์์กด์ฑ (DB ์ฐ๊ฒฐ/๋ฐํ ๋ฑ์ ํ์ฉ)
async def get_db_session():
# 1. ์ง์
์: ๋ฆฌ์์ค ํ ๋น
print("โก [DB] ๋น๋๊ธฐ ์ธ์
์ฐ๊ฒฐ ์์ฑ")
session = {"connected": True}
try:
yield session # ์๋ํฌ์ธํธ ํธ๋ค๋ฌ๋ก ์ธ์
๊ฐ์ฒด ์ ๋ฌ
finally:
# 2. ํธ๋ค๋ฌ ์๋ฃ ๋๋ ์์ธ ๋ฐ์ ์: ๋ฆฌ์์ค ์์ ํด์
print("๐ [DB] ๋น๋๊ธฐ ์ธ์
์ปค๋ฐ/๋กค๋ฐฑ ํ ์ข
๋ฃ")
# [4] ์๋ํฌ์ธํธ์์ ์์กด์ฑ ์ฃผ์
ํ์ฉ
@app.get("/secure-data")
async def get_secure_data(
api_key: Annotated[str, Depends(verify_api_key)],
paging: Annotated[PaginationParams, Depends()],
db: Annotated[dict, Depends(get_db_session)]
):
return {
"status": "authorized",
"page": paging.page,
"offset": paging.offset,
"limit": paging.size
}
```
---
## 7. SQLAlchemy 2.0 ๋น๋๊ธฐ DB ์ฐ๋ (Async Database ORM)
```python
# app/core/database.py
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, DateTime, func, select
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
# PostgreSQL ๋น๋๊ธฐ ์ ์ ์: "postgresql+asyncpg://user:password@localhost:5432/mydb"
# 1. ๋น๋๊ธฐ ์์ง ์์ฑ
engine = create_async_engine(
DATABASE_URL,
echo=False, # ํ๋ก๋์
์์๋ SQL ๋ก๊ทธ๋ฅผ ๋๊ณ , ๋๋ฒ๊ทธ ์ True
pool_size=10, # ์ปค๋ฅ์
ํ ํฌ๊ธฐ (SQLite๋ ๋ฌด์๋จ)
max_overflow=20
)
# 2. ์ธ์
ํฉํ ๋ฆฌ ์์ฑ
AsyncSessionLocal = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
autoflush=False
)
# 3. Base ์ํฐํฐ ๋ชจ๋ธ ์ ์ธ
class Base(DeclarativeBase):
pass
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String(100), nullable=False)
price: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
# 4. FastAPI์ฉ ์ธ์
์ฃผ์
์์กด์ฑ ํจ์
async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit() # ์ ์ ์๋ฃ ์ ์๋ ์ปค๋ฐ
except Exception:
await session.rollback() # ์๋ฌ ๋ฐ์ ์ ์๋ ๋กค๋ฐฑ
raise
```
```python
# ์๋ํฌ์ธํธ์์ ๋น๋๊ธฐ DB ์ฟผ๋ฆฌ ์คํ
from fastapi import APIRouter, Depends
from typing import Annotated
db_router = APIRouter(prefix="/db-items", tags=["Database Items"])
@db_router.get("/")
async def list_items(db: Annotated[AsyncSession, Depends(get_async_db)]):
stmt = select(Item).order_by(Item.id.desc()).limit(10)
result = await db.execute(stmt)
items = result.scalars().all()
return [{"id": item.id, "title": item.title, "price": item.price} for item in items]
@db_router.post("/")
async def create_item(title: str, price: int, db: Annotated[AsyncSession, Depends(get_async_db)]):
new_item = Item(title=title, price=price)
db.add(new_item)
await db.flush() # DB์ ์ ์กํ์ฌ new_item.id ์ฑ๋ฒ
await db.refresh(new_item)
return {"id": new_item.id, "title": new_item.title}
```
---
## 8. ์ธ์ฆ & ๋ณด์ (OAuth2 & JWT Token)
```python
# app/core/security.py
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
SECRET_KEY = "your-super-secret-jwt-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") # Swagger UI์ ์๋ฌผ์ ๋ฒํผ ์๋ ํ์ฑํ
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
# ํ ํฐ ํ์ด๋ก๋ ๊ฒ์ฆ ๋ฐ ์ฌ์ฉ์ ์๋ณ ์์กด์ฑ
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="์๊ฒฉ ์ฆ๋ช
์ ํ์ธํ ์ ์์ต๋๋ค.",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str | None = payload.get("sub")
if username is None:
raise credentials_exception
return {"username": username, "role": payload.get("role", "user")}
except jwt.PyJWTError:
raise credentials_exception
```
```python
# app/api/v1/endpoints/auth.py
from fastapi import APIRouter
auth_router = APIRouter(prefix="/auth", tags=["Auth"])
# 1. ๋ก๊ทธ์ธ ๋ฐ JWT ๋ฐ๊ธ ์๋ํฌ์ธํธ (Swagger ์์ ํธํ)
@auth_router.post("/login")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
# ์ค์ ํ๊ฒฝ: DB์์ form_data.username ์กฐํ ๋ฐ verify_password ๊ฒ์ฆ
if form_data.username != "admin" or form_data.password != "password123":
raise HTTPException(status_code=400, detail="์์ด๋ ๋๋ ๋น๋ฐ๋ฒํธ๊ฐ ์๋ชป๋์์ต๋๋ค.")
access_token = create_access_token(data={"sub": form_data.username, "role": "admin"})
return {"access_token": access_token, "token_type": "bearer"}
# 2. ์ธ์ฆ๋ ์ฌ์ฉ์ ์ ์ฉ ์๋ํฌ์ธํธ
@auth_router.get("/me")
async def get_my_profile(current_user: Annotated[dict, Depends(get_current_user)]):
return {"user": current_user}
```
---
## 9. ๋ฏธ๋ค์จ์ด & CORS & Lifespan ์๋ช
์ฃผ๊ธฐ
```python
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
# [1] ํ๋์ ๋น๋๊ธฐ ์๋ช
์ฃผ๊ธฐ ๋งค๋์ (startup / shutdown ์ด๋ฒคํธ ๋์ฒด)
@asynccontextmanager
async def lifespan(app: FastAPI):
# ์ฑ ๊ตฌ๋(Startup) ์ ์คํ: DB ์ปค๋ฅ์
ํ ์ด๊ธฐํ, ์บ์ ์๋ฐ์
print("๐ [Startup] ๋ฐ์ดํฐ๋ฒ ์ด์ค ์ปค๋ฅ์
ํ ์ด๊ธฐํ ์๋ฃ")
yield
# ์ฑ ์ข
๋ฃ(Shutdown) ์ ์คํ: ์ปค๋ฅ์
ํ ์ข
๋ฃ, ์ธ๋ถ ๋ฆฌ์์ค ์ ๋ฆฌ
print("๐ [Shutdown] ๋ฆฌ์์ค ์์ ํด์ ๋ฐ ์๋ฒ ์ ์ ์ข
๋ฃ")
app = FastAPI(lifespan=lifespan)
# [2] CORS(Cross-Origin Resource Sharing) ์ค์
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000",
"https://myapp.luckytech.com"
],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
allow_headers=["*"],
expose_headers=["X-Process-Time"]
)
# [3] ์๋ต ์์ถ GZip ๋ฏธ๋ค์จ์ด
app.add_middleware(GZipMiddleware, minimum_size=1000)
# [4] ์ปค์คํ
HTTP ์คํ ์๊ฐ ์ธก์ ๋ฏธ๋ค์จ์ด
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
response.headers["X-Process-Time"] = f"{process_time * 1000:.2f}ms"
return response
```
---
## 10. APIRouter ๋ชจ๋ํ & ๋ฐฑ๊ทธ๋ผ์ด๋ ํ์คํฌ (Modularization & Background Tasks)
```python
# app/api/v1/endpoints/notifications.py
from fastapi import APIRouter, BackgroundTasks
import asyncio
router = APIRouter(prefix="/notify", tags=["Notifications"])
# ๋ฌด๊ฑฐ์ด ๋น๋๊ธฐ ์์
ํจ์ (์ด๋ฉ์ผ ๋ฐ์ก, ๊ฐ์ฌ ๋ก๊ทธ ์ ์ก, ํต๊ณ ์ง๊ณ ๋ฑ)
def send_email_task(email: str, message: str):
# ์ค์ ๋ก๋ smtplib, aiosmtplib ๋๋ Celery/Redis Queue ํธ์ถ
print(f"๐ง [Email Sent] ์์ ์: {email}, ๋ณธ๋ฌธ: {message}")
async def write_audit_log(user_id: int, action: str):
await asyncio.sleep(1) # ๋น๋๊ธฐ I/O ์๋ฎฌ๋ ์ด์
print(f"๐ [Audit Log] User {user_id} - Action: {action}")
# BackgroundTasks ์ฃผ์
@router.post("/send")
async def trigger_notification(
email: str,
background_tasks: BackgroundTasks
):
# 1. ์ฆ์ HTTP 202 Accepted ์๋ต ๋ฐํ
# 2. ํด๋ผ์ด์ธํธ ์๋ต ์ ์ก ์๋ฃ ํ ๋ฐฑ๊ทธ๋ผ์ด๋์์ ์์
์์ฐจ ์คํ
background_tasks.add_task(send_email_task, email=email, message="ํ์๊ฐ์
์ ํ์ํฉ๋๋ค!")
background_tasks.add_task(write_audit_log, user_id=42, action="SEND_WELCOME_EMAIL")
return {"status": "accepted", "message": "์๋ฆผ ์์
์ด ๋ฐฑ๊ทธ๋ผ์ด๋์ ๋ฑ๋ก๋์์ต๋๋ค."}
```
```python
# app/main.py ์์ ๋ผ์ฐํฐ ํตํฉ ๋ฑ๋ก
from fastapi import FastAPI
from app.api.v1.endpoints import auth, users, notifications
app = FastAPI()
# API ๋ฒ์ ๋ณ ์ ๋์ฌ(prefix) ๋ฐ ํ๊ทธ ๊ณ์ธต ๋ถ๋ฆฌ
api_v1_router = APIRouter(prefix="/api/v1")
api_v1_router.include_router(auth.auth_router)
api_v1_router.include_router(notifications.router)
app.include_router(api_v1_router)
```
---
## 11. ํ
์คํธ & ํ๋ก๋์
๋ฐฐํฌ (Testing & Docker Deployment)
```python
# tests/test_api.py - FastAPI ๊ณต์ TestClient (HTTPX ๊ธฐ๋ฐ)
import pytest
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_health_check():
response = client.get("/")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_create_user_validation_error():
# ํ์ ํ๋ ๋๋ฝ ์ Pydantic 422 ์ํ ์ฝ๋ ๋ฐํ ํ์ธ
response = client.post("/api/v1/users/", json={"username": "ab"}) # min_length=3 ์๋ฐ
assert response.status_code == 422
assert "detail" in response.json()
```
```dockerfile
# Dockerfile - ๋ฉํฐ ์คํ
์ด์ง ํ๋ก๋์
๊ฒฝ๋ ์ปจํ
์ด๋
FROM python:3.12-slim AS builder
WORKDIR /app
RUN pip install --no-cache-dir uv
COPY pyproject.toml .
RUN uv pip install --system --no-cache "fastapi[standard]" "uvicorn[standard]"
# ์ต์ข
๋ฐํ์ ์ด๋ฏธ์ง
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY ./app ./app
# ๋ณด์์ ์ํ ๋น-๋ฃจํธ ์ฌ์ฉ์ ์คํ
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
# Gunicorn + UvicornWorker๋ฅผ ํ์ฉํ ๋ฉํฐ ํ๋ก์ธ์ค ๊ณ ์ฑ๋ฅ ํ๋ก๋์
์คํ
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4", "--proxy-headers"]
```
์๊ฒฌ ๋ฐ ์ง๋ฌธ
0์์ง ๋ฑ๋ก๋ ์๊ฒฌ์ด ์์ต๋๋ค. ์ฒซ ๋ฒ์งธ ๋๊ธ์ ๋จ๊ฒจ๋ณด์ธ์!
๋๊ธ ์์
๋๊ธ ์ญ์