Today I Learned
오늘 공부한 내용은 FastAPI 2번째 내용!
주간학습정리
학습 정리로 월~금 동안 매일
복습, 과제, 피어세션, 회고 정리했고 아래 링크 달았습니다.
팀 회고
FastAPI - 2
Pydantic
데이터 유효성 검사 및 설정(config) 관리를 위한 라이브러리
Validation
-
기본 타입과 list, dict, tuple에 대한 validation도 가능
-
기존 validation 라이브러리보다 빠르고 머신러닝 Feature 데이터에 대한 Validation도 가능
-
일반 python class로 하면 init 부분과 validation 로직이 길어진다.
dataclass를 활용하면 init은 매직메서드로 생략 가능하고, 인스턴스 생성 시점에 검증할 수 있지만, 여전히 validation 로직이 길다. -
Pydantic을 사용하면 간결하게 검증코드를 작성할 수 있고, http url, enum, db url 같은 validation이 이미 만들어져 있어 사용이 가능하다. 그리고 런타임에서 type hint에 따라 validation error도 잡을 수 있다.
-
기본 모델 정의
from enum import Enum
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl
app = FastAPI()
# Enum 정의
class UserType(str, Enum):
ADMIN = "admin"
USER = "user"
GUEST = "guest"
# 데이터 모델 정의
class Settings(BaseModel):
# HTTP URL 검증
api_url: HttpUrl
# Enum 검증
user_type: UserType
# Float 검증 (최소값, 최대값 설정)
score: float = Field(ge=0.0, le=100.0)
@app.post("/settings/")
async def create_settings(settings: Settings):
return settingsConfig 관리
-
코드 상에서 상수로 관리할 수도 있지만 보안 이슈가 있다.
YAML 파일로 관리할 때는 배포 환경별로 파일 생성해서 관리한다. -
Pydantic.BaseSetting을 사용하면 환경변수에 설정값을 저장하고 코드에선 그 값을 읽어와서 작동한다. 배포할때 환경변수를 주입해서 사용한다.
-
보통 .env-example 파일을 만들어 이걸 수정해서 사용한다.
from pydantic import BaseSettings, Field
from enum import Enum
class DatabaseSettings(BaseSettings):
host: str = Field(default='localhost', env='DB_HOST')
port: int = Field(default=5432, env='DB_PORT')
username: str = Field(default='user', env='DB_USERNAME')
password: str = Field(env='DB_PASSWORD')
# 데이터베이스 설정
database: DatabaseSettings = DatabaseSettings()
class Config:
env_file = '.env'
env_file_encoding = 'utf-8'
lifespan function
-
FastAPI에서 lifecycle을 다루는 함수로 앱을 실행할 때나 종료할 때 로직을 넣을 경우 사용
ex) 앱 켜지고 모델 로드, 앱 꺼질때 DB 커넥션 정리 -
@asynccontextmanager 로 지정한 함수 안에서 yield 이전은 앱 시작 전, yield 이후는 앱 종료 후의 작업을 지정한다.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from transformers import AutoModel
# 전역 변수로 모델과 DB 엔진 선언
model = None
engine = None
@asynccontextmanager
async def lifespan(app: FastAPI):
# 시작 시 실행되는 코드
global model, engine
# ML 모델 로딩
print("Loading ML model...")
model = AutoModel.from_pretrained("bert-base-uncased")
# DB 엔진 생성
engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/dbname",
echo=True
)
yield # FastAPI 애플리케이션 실행
# 종료 시 실행되는 코드
print("Cleaning up DB connections...")
if engine:
await engine.dispose()
print("Cleaning up model resources...")
if model:
del model
# FastAPI 앱 생성. 파라미터로 lifespan 주입
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict():
# 로드된 모델 사용
return {"model_loaded": model is not None}
@app.get("/db-status")
async def db_status():
# DB 연결 상태 확인
return {"database_connected": engine is not None}API Router
FastAPI에서 제공하는 라우팅 모듈화 도구로, 대규모 애플리케이션의 구조화와 관리를 용이하게 만든다.
- API Router 사용시 프로젝트 구조
app/
├── api/
│ ├── users.py
│ ├── items.py
│ └── admin.py
├── core/
└── main.py
- users.py
from fastapi import FastAPI, APIRouter
import uvicorn
user_router = APIRouter(prefix="/users")
@user_router.get("/", tags=["users"])
def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
@user_router.get("/me", tags=["users"])
def read_user_me():
return {"username": "fakecurrentuser"}
app = FastAPI()
if __name__ == '__main__':
app.include_router(user_router)
uvicorn.run(app, host="0.0.0.0", port=8000)
Background Tasks
-
오래 기다려야되는 작업은 비동기로 background에서 작업하는게 좋다.
-
async def로 함수 정의를 하고 파라미터로 background_tasks: BackgroundTasks를 준 뒤 background_tasks.add_task() 함수로 작업을 처리한다.
-
이메일 전송 예시
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
async def send_welcome_email(user_email: str):
# 이메일 발송 로직
pass
@app.post("/signup")
async def signup(user_email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(send_welcome_email, user_email)
return {"message": "Signup successful"}Test
-
TestClient를 사용하여 FastAPI 애플리케이션을 테스트
-
test_main.py
from fastapi.testclient import TestClient
from main import app
import pytest
client = TestClient(app)
def test_read_item():
response = client.get("/items/1")
assert response.status_code == 200
assert response.json() == {"item_id": 1}
