
main
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
- async def: FastAPI의 핵심. 입출력(I/O) 대기 시간 동안 다른 작업을 처리할 수 있게 해주는 비동기 문법.
- 데코레이터(@): HTTP 메서드(GET, POST, PUT, DELETE)와 URL 경로를 함수와 연결
endpoints
router = APIRouter()
@router.get(
"/users",
response_model=UserListResponse,
summary="[관리자] 사용자 목록 조회",
description="관리자 권한으로 전체 사용자 목록을 검색 및 필터링하여 조회합니다."
)
async def list_users(
# 1. 검색
search: Optional[str] = Query(None, description="검색어 (이름 또는 이메일)"),
# 2. 페이징
page: int = Query(1, ge=1, description="페이지 번호"),
page_size: int = Query(20, ge=1, le=100, description="페이지당 결과 수"),
# 3. 의존성 주입
_admin: User = Depends(require_admin), # 관리자 권한 체크 (문지기 역할)
db: AsyncSession = Depends(get_db), # 비동기 DB 세션 주입
) -> UserListResponse:
"""
관리자 페이지의 사용자 관리 테이블을 위한 API입니다.
"""
# 서비스 로직 호출 (예시)
# users = await user_service.get_all(db, search, status, page, page_size)
# return users
pass
- Query()를 통해 가능한 것
- 기본값 설정: Query(None)은 값이 안 들어오면 None을 넣겠다는 뜻
- 유효성 검사: Query(None, min_length=3)
- Swagger 문서 설명 추가: Query(..., description="설명")
- Depends(함수/클래스): 의존성 주입
- 스프링 부트와 달리 FastAPI는 요청(Request)이 들어올 때마다 주입이 실행되는 경우가 많음(캐싱은 됨)
- 해당 함수를 실행하기 전에 미리 필요한 함수나 클래스, callable 객체를 파라미터에 꽂아줌
- 파라미터로 함수가 들어가면 그 함수를 실행한 후 결과를 파라미터에 꽂아줌
- 파라미터로 클래스가 들어가면 그 클래스의 인스턴스를 파라미터에 꽂아줌 (스프링 부트처럼 상태관리 로직을 가진 Service 클래스 등)
- 아래 get_db 예시
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Repositories
class UserRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def list_for_admin(
self,
*,
search: str | None,
page: int,
page_size: int,
) -> tuple[list[tuple[User, datetime | None]], int]:
- self 파라미터: 파이썬 클래스 내부에 정의된 함수(메서드)는 첫 번째 인자로 무조건 인스턴스 자신을 받아야 함.
- 클래스 안에 있는 다른 변수나 함수에 접근할 수 있게 해줌
- 파이썬이 내부적으로 첫 번째 인자에 self를 자동으로 넣어 줌
- 만약 이 클래스 안에 self.db가 저장되어 있다면, list_for_admin 내부에서 self.db를 꺼내어 데이터베이스 쿼리를 날릴 수 있음
- * (Keyword-Only Arguments): 이 뒤에 오는 모든 인자는 반드시 이름을 지정해서 넘겨야 한다"는 강제 규칙
- *가 없을 때: service.list_for_admin(None, "ALL", ...) (순서대로 막 넣음)
- *가 있을 때: service.list_for_admin(search=None, search_key="ALL" ...)
- tuple: 순서가 있는 요소들의 집합으로, 한 번 생성되면 값을 변경, 추가, 삭제할 수 없는 불변 자료구조
- 함수 반환값으로 튜플을 사용해 결과값과 전체 카운트를 동시에 넘겨주는 것은 파이썬 백엔드의 전형적인 관례
- users, total_count = await service.list_for_admin(...) 처럼 사용.
from sqlalchemy import select, func
from sqlalchemy.orm import Session
# 1. 서브쿼리 정의: 유저별로 가장 최근 게시글의 업데이트 시간을 구함
last_post_subq = (
select(
Post.user_id.label("user_id"),
func.max(Post.updated_at).label("latest_post_at"),
)
.where(Post.is_deleted == False) # 삭제되지 않은 게시글만
.group_by(Post.user_id) # 유저별로 그룹핑
.subquery() # 이 쿼리를 하나의 가상 테이블(서브쿼리)로 만듦
)
# 2. 메인 쿼리: 유저 정보와 서브쿼리의 결과를 합침(JOIN)
stmt = (
select(
User,
last_post_subq.c.latest_post_at # 서브쿼리의 컬럼은 .c (column)로 접근
)
.outerjoin(
last_post_subq,
User.id == last_post_subq.c.user_id
)
)
# 결과 실행 예시:
# result = await db.execute(stmt)
# for user, last_date in result:
# print(f"{user.name}님의 마지막 포스팅: {last_date}")
- SQLAlchemy 2.0: SQL 문법을 파이썬 함수 형태로 작성하게 해 주는 ORM
- 비동기 환경에 최적화됨
- Model(위 코드에서는 User)을 가지고 있어서 타입 에러나 컬럼 오류를 미리 잡을 수 있음
Schemas
from pydantic import BaseModel, Field
class AccountRestriction(BaseModel):
"""사용자 정지 정보 (정지 상태인 유저에게만 포함되는 상세 정보)"""
reason: RestrictionReason = Field(..., description="정지 사유")
duration_days: int = Field(
...,
ge=1,
le=365,
description="정지 총 기간 (단위: 일)"
)
until: datetime = Field(
...,
description="정지 해제 예정 일시 (ISO 8601 형식)"
)
- Pydantic: 파이썬의 타입 힌트를 이용해서 데이터를 검증하고 설정을 관리하는 가장 빠르고 정확한 라이브러리
- 클라이언트가 보낸 데이터가 모델 형식과 맞지 않으면, 422 Unprocessable Entity 에러를 내려줌
- BaseModel: Pydantic의 핵심 클래스. 스프링으로 치면 Getter/Setter, Validation, JSON 직렬화 기능 추가
- Field: Query와 비슷하게 기본값 설정, 유효성 검사, Swagger 설명 추가 가능.
- Query는 '쿼리' 파라미터를 정의할 때, Field는 클래스의 내부 '필드'를 정의할 때 사용.
- .model_dump()나 .model_dump_json() 같은 메서드를 통해 쉽게 딕셔너리나 JSON 문자열로 변환됨
- 이걸 DB 모델에 바로 kwargs로 쏟아부을 수 있어서 toEntity() 같은 메서드 잘 안씀
- user = User(**filters.model_dump())
- 이걸 DB 모델에 바로 kwargs로 쏟아부을 수 있어서 toEntity() 같은 메서드 잘 안씀
기타 문법 및 로직
- 파이썬 3.10부터 switch문 비슷한게 추가되었음(match status 문)
def get_status_message(status):
match status:
case "ACTIVE":
return "활동 중"
case "BANNED":
return "정지됨"
case "PENDING" | "INACTIVE": # 여러 조건을 하나로 묶을 수도 있음!
return "대기 또는 비활성"
case _: # 다른 언어의 'default' 역할
return "알 수 없는 상태"
- Enum을 정의할 수 있음
from enum import Enum
class UserGrade(str, Enum):
"""서비스 내부에서 사용하는 표준 등급"""
BRONZE = "BRONZE"
SILVER = "SILVER"
GOLD = "GOLD"
ADMIN = "ADMIN"
- 파이썬 딕셔너리의 get() 메서드를 이용한 Enum 매핑 폴백(fallback) 로직
# 외부 시스템(예: 구버전 DB) ↔ 우리 서비스 Enum 매핑
_EXTERNAL_TO_INTERNAL_GRADE: dict[str, UserGrade] = {
"common": UserGrade.BRONZE, # 외부 시스템의 'common'은 우리 서비스의 'BRONZE'
"vip": UserGrade.SILVER,
"vvip": UserGrade.GOLD,
}
def get_grade(raw_value: str | None) -> UserGrade:
"""
DB에서 가져온 날것의 값을 우리 Enum으로 변환.
이상한 값이 들어오면 가장 낮은 등급(BRONZE)으로 회귀(Fallback).
"""
return _EXTERNAL_TO_INTERNAL_GRADE.get(raw_value, UserGrade.BRONZE)
맨 아랫줄의 get() 을 보면 두 번째 인자에 BRONZE가 들어가 있다. fallback할 값이 들어갈 위치이다.
- Depends()로 가독성 향상
class PaginationParams:
def __init__(self, page: int = Query(1, ge=1), size: int = Query(20, le=100)):
self.page = page
self.size = size
@router.get("/items")
async def read_items(params: PaginationParams = Depends(PaginationParams)):
# URL의 ?page=1&size=20을 읽어서 params 객체로 만들어줌
return {"page": params.page, "size": params.size}
모든 API마다 page, size 파라미터를 주렁주렁 달기 귀찮으니, DTO처럼 PaginationParams라는 클래스 하나로 묶어서 관리
슬슬 백엔드 다룰 일이 많아질 것 같아서 FastAPI 문법을 정리해봤습니다.
스프링 부트랑 비슷하면서도 다른 점이 많고..
확실히 파이썬은 라이브러리 하나하나가 강력한 느낌?
'IT > Python' 카테고리의 다른 글
| [Python] BFS (너비 우선 탐색) 알고리즘 (0) | 2024.06.16 |
|---|---|
| [Python] DFS(깊이우선탐색) 알고리즘 (1) | 2024.01.28 |
| [Python] 지역변수와 전역변수 (0) | 2024.01.18 |
| [Python] 최소힙(heap) 자료구조 (1) | 2024.01.08 |
| [알고리즘 공부] 그리디(Greedy) 알고리즘 (1) | 2023.10.31 |