122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""
|
|
Pagination Dependencies
|
|
|
|
페이지네이션 관련 의존성 주입을 정의합니다.
|
|
|
|
사용 예시:
|
|
from app.dependencies.pagination import PaginationParams, get_pagination_params, paginated_query
|
|
|
|
# 방법 1: 기본 파라미터만 사용
|
|
@router.get("/items", response_model=PaginatedResponse[ItemModel])
|
|
async def get_items(
|
|
pagination: PaginationParams = Depends(get_pagination_params),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
...
|
|
|
|
# 방법 2: paginated_query 사용 (get_paginated 래핑)
|
|
@router.get("/items", response_model=PaginatedResponse[ItemModel])
|
|
async def get_items(
|
|
session: AsyncSession = Depends(get_session),
|
|
pagination: PaginationParams = Depends(get_pagination_params),
|
|
):
|
|
return await paginated_query(
|
|
session=session,
|
|
model=Item,
|
|
item_schema=ItemModel,
|
|
pagination=pagination,
|
|
filters={"status": "completed"},
|
|
)
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Annotated, Any, Callable, Dict, Optional, Type, TypeVar
|
|
|
|
from fastapi import Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.utils.pagination import PaginatedResponse, get_paginated
|
|
|
|
T = TypeVar("T")
|
|
ModelT = TypeVar("ModelT")
|
|
|
|
|
|
@dataclass
|
|
class PaginationParams:
|
|
"""페이지네이션 파라미터를 담는 데이터 클래스"""
|
|
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
def get_pagination_params(
|
|
page: Annotated[int, Query(ge=1, description="페이지 번호 (1부터 시작)")] = 1,
|
|
page_size: Annotated[
|
|
int, Query(ge=1, le=100, description="페이지당 데이터 수 (최대 100)")
|
|
] = 10,
|
|
) -> PaginationParams:
|
|
"""페이지네이션 파라미터를 주입하는 의존성 함수
|
|
|
|
Args:
|
|
page: 페이지 번호 (1부터 시작, 기본값: 1)
|
|
page_size: 페이지당 데이터 수 (기본값: 10, 최대: 100)
|
|
|
|
Returns:
|
|
PaginationParams: 페이지네이션 파라미터 객체
|
|
"""
|
|
return PaginationParams(page=page, page_size=page_size)
|
|
|
|
|
|
async def paginated_query(
|
|
session: AsyncSession,
|
|
model: Type[ModelT],
|
|
item_schema: Type[T],
|
|
pagination: PaginationParams,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
order_by: Optional[str] = "created_at",
|
|
order_desc: bool = True,
|
|
transform_fn: Optional[Callable[[ModelT], T]] = None,
|
|
) -> PaginatedResponse[T]:
|
|
"""페이지네이션 쿼리를 실행하는 헬퍼 함수
|
|
|
|
PaginationParams를 받아서 get_paginated를 호출합니다.
|
|
|
|
Args:
|
|
session: SQLAlchemy AsyncSession
|
|
model: SQLAlchemy 모델 클래스 (예: Song, Lyric, Video)
|
|
item_schema: Pydantic 스키마 클래스 (예: SongListItem)
|
|
pagination: 페이지네이션 파라미터 (get_pagination_params에서 주입)
|
|
filters: 필터 조건 딕셔너리 (예: {"status": "completed"})
|
|
order_by: 정렬 기준 컬럼명 (기본값: "created_at")
|
|
order_desc: 내림차순 정렬 여부 (기본값: True)
|
|
transform_fn: 모델을 스키마로 변환하는 함수 (None이면 자동 변환)
|
|
|
|
Returns:
|
|
PaginatedResponse[T]: 페이지네이션된 응답
|
|
|
|
Usage:
|
|
@router.get("/songs", response_model=PaginatedResponse[SongListItem])
|
|
async def get_songs(
|
|
session: AsyncSession = Depends(get_session),
|
|
pagination: PaginationParams = Depends(get_pagination_params),
|
|
):
|
|
return await paginated_query(
|
|
session=session,
|
|
model=Song,
|
|
item_schema=SongListItem,
|
|
pagination=pagination,
|
|
filters={"status": "completed"},
|
|
)
|
|
"""
|
|
return await get_paginated(
|
|
session=session,
|
|
model=model,
|
|
item_schema=item_schema,
|
|
page=pagination.page,
|
|
page_size=pagination.page_size,
|
|
filters=filters,
|
|
order_by=order_by,
|
|
order_desc=order_desc,
|
|
transform_fn=transform_fn,
|
|
)
|