기능이 늘어도 버티는 구조는 라우팅과 업무 로직과 공통 설정을 갈라 두는 것이다. 한 파일에 다 넣으면 엔드포인트가 열 개를 넘는 순간 손대기 어려워진다.
noti_api/
├── main.py # 앱 생성과 라우터 등록만 한다
├── config.yaml # 런타임 설정
├── requirements.txt
├── app/
│ ├── core/
│ │ ├── config.py # 설정 로딩과 기본값
│ │ └── logger.py # 로깅 초기화
│ ├── routes/
│ │ ├── health.py # /health
│ │ └── message.py # /v1/message
│ └── services/
│ ├── mailer.py # SMTP 발송
│ └── storage.py # 파일·객체 저장
└── deploy/
├── noti-api.service
└── Dockerfile
routes 는 요청을 받아 검증하고 services 를 부르는 데까지만 한다. 업무 로직을 라우트 함수 안에 쓰면 테스트할 수 없다.
Python 3.9 에서도 FastAPI 와 Uvicorn 이 동작한다. 다만 3.9 는 보안 지원이 끝나는 계열이므로 가능하면 3.11 이상으로 올린다.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.core.config import load_config
from app.core.logger import setup_logging
from app.routes import health, message
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.config = load_config()
setup_logging(app.state.config)
# 기동 시 한 번 하는 일: 커넥션 풀 생성, 캐시 준비
yield
# 종료 시 정리: 풀 close, 임시 파일 제거
app = FastAPI(title="noti-api", lifespan=lifespan)
app.include_router(health.router)
app.include_router(message.router, prefix="/v1")
lifespan 은 예전의 on_event startup · shutdown 을 대체한다. 두 이벤트가 하나의 컨텍스트 매니저로 합쳐져 yield 앞이 기동, 뒤가 종료다. 기동 때 연 자원을 종료 때 닫는 짝을 한 함수 안에서 보게 되므로 빠뜨리기 어렵다. on_event 방식은 사용 중단 상태다.
운영 실행은 워커를 여럿 두고 프로세스 관리를 Gunicorn 에 맡기는 구성이 무난하다.
gunicorn main:app \
-k uvicorn.workers.UvicornWorker \
-w 4 -b 127.0.0.1:8000 \
--timeout 60 --graceful-timeout 30 --access-logfile -
워커 수는 CPU 코어 수를 기준으로 잡되, 대부분의 시간을 외부 API 응답 대기로 보내는 서비스라면 코어 수보다 많이 두어도 된다. 동기 함수로 블로킹 호출을 하면 이벤트 루프가 멈추므로, 동기 라이브러리를 쓸 때는 라우트를 def 로 선언해 FastAPI 가 스레드풀로 돌리게 한다.
[Unit]
Description=noti-api
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=apiuser
Group=apiuser
WorkingDirectory=/opt/noti_api
Environment=NOTI_CONFIG=/opt/noti_api/config.yaml
EnvironmentFile=-/etc/sysconfig/noti-api
ExecStart=/opt/noti_api/.venv/bin/gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 127.0.0.1:8000
Restart=on-failure
RestartSec=5
NoNewPrivileges=true
PrivateTmp=true
[Install]
WantedBy=multi-user.target
systemctl daemon-reload
systemctl enable --now noti-api
journalctl -u noti-api -f
앞단에는 Nginx 를 두어 TLS 종료와 정적 파일을 맡긴다. 프록시 헤더를 넘기고, 애플리케이션은 forwarded-allow-ips 옵션으로 그것을 신뢰하도록 설정한다.
두 종류를 나눈다. 프로세스가 살아 있는지만 보는 것과, 의존하는 외부 자원까지 확인하는 것이다.
@router.get("/health") # 살아 있는가. 항상 가볍게
def health():
return {"status": "ok"}
@router.get("/ready") # 일할 준비가 됐는가
def ready():
checks = {"db": check_db(), "smtp": check_smtp()}
ok = all(checks.values())
return JSONResponse(
{"status": "ok" if ok else "degraded", "checks": checks},
status_code=200 if ok else 503,
)
로드밸런서가 health 를 보게 하고, 배포 시점 판단에는 ready 를 쓴다. 헬스체크가 외부 자원을 매번 호출하면 그 자원이 흔들릴 때 전체가 함께 내려간다.
로그 수준을 설정으로 고를 수 있게 하고, 요청 단위로 상관 ID 를 남긴다.
import logging
import logging.config
def setup_logging(cfg):
level = cfg.get("log", {}).get("level", "INFO").upper()
logging.config.dictConfig({
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"std": {"format": "%(asctime)s %(levelname)s [%(name)s] %(message)s"},
},
"handlers": {
"console": {"class": "logging.StreamHandler", "formatter": "std"},
},
"root": {"level": level, "handlers": ["console"]},
})
systemd 로 돌리면 표준 출력이 journald 로 간다. 파일로 남길 이유가 없으면 그대로 두고 수집기로 보낸다. 파일로 남긴다면 RotatingFileHandler 를 쓰되, 워커가 여럿이면 같은 파일에 동시에 써서 로테이션이 깨진다. 이 경우 logrotate 의 copytruncate 를 쓰거나 표준 출력으로 내보낸다.
DEBUG 수준에서 요청 본문을 통째로 남기면 개인정보와 토큰이 함께 남는다. 운영 기본은 INFO 로 두고 필요할 때만 한시적으로 올린다.