fastapi + sqlalchemy + ORM을 이용한 페이지네이션 방법들

1. Offset/Limit 방식

   - skip : 몇 개의 항목을 건너뛸지 결정한다.
   - limit : 가져올 항목의 최대 개수를 설정한다.

from fastapi import FastAPI, Query, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import List, Any
from .database import get_db
from .models import Item

app = FastAPI()

# Response schema for consistent structure
class PaginatedResponse(BaseModel):
    total: int  # Total number of items
    skip: int  # Current skip value
    limit: int  # Current limit value
    data: List[Any]  # List of items

@app.get("/items/", response_model=PaginatedResponse)
def read_items(
    skip: int = Query(0, ge=0, description="Number of items to skip"),
    limit: int = Query(10, ge=1, le=100, description="Number of items to fetch"),
    db: Session = Depends(get_db)
):
    # Count total items for pagination
    total = db.query(Item).count()

    # Retrieve paginated items
    items = db.query(Item).offset(skip).limit(limit).all()

    # Return a structured response
    return {
        "total": total,
        "skip": skip,
        "limit": limit,
        "data": items,
    }

 

2. Cursor 기반 페이지네이션

   - Cursor 기반 페이지네이션은 고유한 ID를 기준으로 다음 페이지의 시작점을 결정한다.

   - cursor : 마지막으로 가져온 항목의 ID를 기준으로 다음 페이지를 가져온다.

from fastapi import FastAPI, Query, Depends
from sqlalchemy.orm import Session
from .database import get_db
from .models import Item

app = FastAPI()

@app.get("/items/")
def read_items(cursor: int = Query(None), limit: int = 10, db: Session = Depends(get_db)):
    query = db.query(Item)
    if cursor:
        query = query.filter(Item.id > cursor)
    items = query.order_by(Item.id).limit(limit).all()
    return items

 

3. SQLAlchemy Paginator 라이브러리 사용

   - SQLAlchemy Paginator와 같은 라이브러리를 사용하여 페이지네이션을 간편하게 처리할 수 있다.

   - page : 가져올 페이지 번호를 설정한다.
   - size : 페이지당 항목 수를 설정한다.

from fastapi import FastAPI, Query, Depends
from sqlalchemy.orm import Session
from sqlalchemy_pagination import paginate
from .database import get_db
from .models import Item

app = FastAPI()

@app.get("/items/")
def read_items(
    page: int = Query(1, description="가져올 페이지 번호 (1부터 시작)"),
    size: int = Query(10, description="페이지당 항목 수"),
    db: Session = Depends(get_db),
):
    """
    항목 목록을 페이지네이션하여 반환합니다.

    Args:
        page (int): 가져올 페이지 번호입니다.
        size (int): 페이지당 항목 수입니다.
        db (Session): 데이터베이스 세션 객체입니다.

    Returns:
        dict: 페이지네이션된 결과를 포함한 객체입니다.
    """
    query = db.query(Item)
    result = paginate(query, page, size)
    return {
        "total": result.total,
        "pages": result.pages,
        "current_page": result.page,
        "items": result.items,
    }