DRF와 FastAPI에서 다양한 페이지네이션 전략 구현하기

소개: 효율적인 페이지네이션으로 대규모 데이터셋 탐색하기

현대 웹 개발에서 방대한 양의 데이터를 처리하는 것은 흔한 과제입니다. API를 통해 리소스 컬렉션을 노출할 때, 전체 데이터셋을 단일 응답으로 반환하는 것은 불가능하지는 않더라도 비실용적인 경우가 많습니다. 이러한 접근 방식은 느린 응답 시간, 서버와 클라이언트 모두의 과도한 메모리 소비, 그리고 열악한 사용자 경험으로 이어질 수 있습니다. 페이지네이션은 데이터를 관리 가능한 덩어리로 검색할 수 있도록 하는 필수적인 솔루션으로 등장합니다. 데이터를 페이지로 나누는 개념은 간단해 보이지만, 다양한 페이지네이션 전략은 다양한 사용 사례에 맞춰 서로 다른 장점과 단점을 제공합니다. 이 글에서는 두 가지 주요 페이지네이션 기법인 Limit/Offset 및 Cursor 기반 페이지네이션을 살펴보고, 두 가지 인기 있는 Python 웹 프레임워크인 Django Rest Framework(DRF)와 FastAPI 내에서의 구현을 시연할 것입니다. 이러한 방법을 이해하는 것은 대규모 데이터셋을 효과적으로 제공할 수 있는 확장 가능하고 견고한 API를 구축하는 데 중요합니다.

핵심 페이지네이션 개념: 입문

구현 세부 사항을 알아보기 전에 페이지네이션 전략의 근간을 이루는 기본 개념을 명확히 하겠습니다.

• 페이지네이션: 대규모 데이터셋을 작고 개별적인 페이지 또는 덩어리로 나누어 클라이언트에게 순차적으로 제공하는 프로세스입니다. 이는 성능을 향상시키고 리소스 사용량을 관리합니다.
• 페이지: 전체 데이터의 하위 집합으로, 일반적으로 크기(페이지당 항목 수)와 식별자(페이지 번호, 오프셋 또는 커서)로 정의됩니다.
• Limit: 단일 응답에서 반환할 최대 항목 수(즉, 페이지 크기)를 나타냅니다.
• Offset: 결과를 반환하기 전에 데이터셋 시작 부분에서 건너뛸 항목 수를 나타냅니다.
• 커서(Cursor): 데이터셋 내의 특정 항목을 가리키는 불투명한 문자열 또는 값입니다. 이는 절대적인 위치(오프셋)에 의존하지 않고 해당 지점에 상대적인 "다음" 또는 "이전" 항목 집합을 검색하기 위한 북마크로 사용됩니다.
• 안정적인 페이지네이션(Stable Pagination): 클라이언트가 페이지네이션하는 동안 데이터셋에 항목이 추가되거나 제거되어도 항목이 건너뛰거나 페이지 간에 중복되지 않는 경우 페이지네이션 전략은 안정적이라고 간주됩니다.

Limit/Offset 페이지네이션: 단순함과 그 함정

Limit/Offset은 아마도 가장 일반적이고 직관적인 페이지네이션 전략일 것입니다. limit(반환할 항목 수)와 offset(건너뛸 항목 수)이라는 두 가지 매개변수를 지정하여 작동합니다.

작동 방식:

클라이언트는 limit와 offset을 제공하여 데이터를 요청합니다. 그런 다음 서버는 offset 번째 레코드부터 시작하여 limit 개의 항목을 가져옵니다. 예를 들어, 페이지당 10개 항목으로 두 번째 페이지를 가져오려면 클라이언트는 limit=10&offset=10을 요청합니다.

장점:

• 단순성: 서버와 클라이언트 모두 이해하고 구현하기 쉽습니다.
• 직접 액세스: 클라이언트는 offset(offset = (page_number - 1) * limit)을 계산하여 어떤 특정 페이지로든 쉽게 이동할 수 있습니다.

단점:

• 큰 오프셋으로 인한 성능 저하: offset이 증가함에 따라 데이터베이스는 여전히 건너뛴 모든 레코드를 스캔해야 할 수 있으며, 특히 적절한 인덱싱이 없는 대규모 테이블에서는 성능 병목 현상이 발생할 수 있습니다.
• 불안정성(항목 건너뛰기/중복): 클라이언트가 페이지네이션하는 동안 현재 오프셋 이전의 데이터셋에 항목이 추가되거나 삭제되면 결과가 일관되지 않을 수 있습니다. 항목이 두 페이지에 나타나거나 완전히 건너뛸 수 있습니다. 제품 목록을 고려하십시오. 사용자가 5페이지를 보고 있는 동안 목록 시작 부분에 새 제품이 추가되면 후속 페이지에 이미 본 항목이 포함되거나 새 항목이 건너뛸 수 있습니다.

DRF에서 Limit/Offset 구현하기

DRF는 내장된 LimitOffsetPagination 클래스를 제공하여 구현을 간단하게 만듭니다.

# project/settings.py
REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination',
    'PAGE_SIZE': 10 # 기본 페이지 크기
}

# app/views.py
from rest_framework import generics
from .models import Product
from .serializers import ProductSerializer

class ProductListView(generics.ListAPIView):
    queryset = Product.objects.all().order_by('id') # 일관된 페이지네이션을 위해 항상 정렬
    serializer_class = ProductSerializer
    # pagination_class = LimitOffsetPagination # 뷰별로 설정할 수도 있습니다.

클라이언트는 /products/?limit=5&offset=10과 같은 요청을 하게 됩니다. PAGE_SIZE를 사용하기 위해 limit을 생략할 수 있습니다.

FastAPI에서 Limit/Offset 구현하기

FastAPI는 더 미니멀리스트적인 프레임워크로서, Pydantic과 의존성을 활용하여 수동 설정이 필요합니다.

# main.py
from typing import List, Optional
from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# 데이터베이스 설정 (예제를 위해 단순화)
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class ProductModel(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)

Base.metadata.create_all(bind=engine)

class ProductCreate(BaseModel):
    name: str
    description: str

class Product(ProductCreate):
    id: int

    class Config:
        orm_mode = True

app = FastAPI()

# DB 세션을 얻기 위한 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# LimitOffset 페이지네이션 의존성
class LimitOffsetParams:
    def __init__(
        self,
        limit: int = Query(10, ge=1, le=100),
        offset: int = Query(0, ge=0),
    ):
        self.limit = limit
        self.offset = offset

@app.post("/products/", response_model=Product)
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    db_product = ProductModel(**product.dict())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

@app.get("/products/", response_model=List[Product])
def get_products(
    pagination: LimitOffsetParams = Depends(),
    db: Session = Depends(get_db)
):
    products = db.query(ProductModel).offset(pagination.offset).limit(pagination.limit).all()
    return products

이 FastAPI 예제에서 LimitOffsetParams는 limit 및 offset 매개변수를 라우트 함수에 직접 주입하는 의존성 역할을 합니다. SQL 쿼리는 데이터를 검색하기 위해 .offset() 및 .limit()을 사용합니다.

Cursor 기반 페이지네이션: 안정성과 성능 보장

Cursor 기반 페이지네이션(Keyset 페이지네이션이라고도 함)은 Limit/Offset의 안정성과 성능 문제를 해결하며, 특히 대규모 데이터셋의 경우 더욱 그렇습니다. 숫자 오프셋 대신, 다음 결과 집합을 검색하기 위한 "마지막으로 본 항목" 포인터(커서)를 사용합니다.

작동 방식:

클라이언트는 페이지네이션된 데이터와 함께 커서 값(일반적으로 ID 또는 타임스탬프와 같은 인코딩된 식별자)을 받습니다. 다음 페이지를 가져오려면 클라이언트는 이 커서를 서버로 다시 보내고, 서버는 해당 커서 값 이후의 항목을 검색합니다. 이는 일관되게 정렬된 데이터에 크게 의존합니다. 예를 들어, ID X 이후의 항목을 가져오려면 쿼리는 WHERE id > X ORDER BY id LIMIT N이 됩니다.

장점:

• 안정성: 페이지네이션 중에 항목이 추가되거나 제거되어도 정렬 순서가 일관되면 후속 페이지에 포함되는 항목에 영향을 미치지 않습니다. 이는 레코드 건너뛰기 또는 중복을 방지합니다.
• 성능: 데이터베이스는 정렬된 열(예: id 또는 timestamp)의 인덱스를 효율적으로 사용하여 시작점을 빠르게 찾을 수 있으므로 큰 오프셋과 관련된 느린 스캔을 피할 수 있습니다. 매우 큰 데이터셋의 경우 훨씬 더 잘 확장됩니다.
• 확장성: 사용자가 일반적으로 한 번에 한 페이지씩만 앞뒤로 이동하는 무한 스크롤 피드 또는 타임라인에 더 적합합니다.

단점:

• 직접 페이지 액세스 불가: 임의의 페이지(예: 5페이지)로 "점프"할 수 있는 숫자 페이지 개념이 없으므로 클라이언트가 할 수 없습니다. 현재 커서에 상대적으로만 이동할 수 있습니다.
• 안정적인 정렬 키 필요: 커서 역할을 할 고유하고 불변하며 순차적으로 정렬할 수 있는 열(기본 키 또는 타임스탬프와 같은)이 필요합니다.
• 뒤로 페이지네이션 복잡성: 뒤로 페이지네이션(예: "이전 페이지")을 구현하는 것은 정렬 및 필터 조건을 반대로 하는 추가 로직이 필요하여 더 복잡할 수 있습니다.

DRF에서 Cursor 기반 페이지네이션 구현하기

DRF는 커서 값의 인코딩/디코딩을 스마트하게 처리하는 CursorPagination을 제공합니다.

# project/settings.py
# 기본값으로 사용하려면
# REST_FRAMEWORK = {
#     'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.CursorPagination',
#     'PAGE_SIZE': 10,
#     'CURSOR_PAGINATION_USE_REL_LINK_HEADERS': True # HATEOAS 링크의 경우 선택 사항
# }

# app/views.py
from rest_framework import generics
from rest_framework.pagination import CursorPagination
from .models import Product
from .serializers import ProductSerializer

# 특정 정렬을 위한 사용자 지정 CursorPagination
class ProductCursorPagination(CursorPagination):
    page_size = 10
    ordering = 'created_at' # 또는 'id', 'name' 등. 고유하고 일관되게 정렬되어야 합니다.
    # cursor_query_param = 'cursor' # 기본값, 변경 가능
    # page_size_query_param = 'page_size' # 기본값, 변경 가능

class ProductListView(generics.ListAPIView):
    queryset = Product.objects.all().order_by('created_at', 'id') # 안정성을 위해 중요
    serializer_class = ProductSerializer
    pagination_class = ProductCursorPagination

ProductCursorPagination의 ordering 속성은 중요합니다. 이는 커서에 사용되는 열과 필요한 정렬 순서를 정의합니다. created_at과 같이 기본 정렬 필드가 고유하지 않은 경우를 처리하기 위해 ordering에 id와 같은 보조 고유 필드를 포함하는 것이 좋은 습관입니다. 요청은 다음 페이지에 대해 /products/?cursor=AbcD...와 같은 형태가 될 것입니다. 여기서 AbcD...는 이전 응답에서 제공된 불투명한 커서 문자열입니다.

FastAPI에서 Cursor 기반 페이지네이션 구현하기

FastAPI에서 cursor 기반 페이지네이션을 구현하려면 사용자 지정 의존성과 쿼리 로직의 신중한 처리가 필요합니다.

# main.py (이전 FastAPI 예제를 기반으로 함)
import base64
from typing import List, Optional
from fastapi import FastAPI, Depends, Query, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.sql import func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from datetime import datetime

# (데이터베이스 설정 및 ProductModel/ProductCreate/Product는 이전과 동일)
class ProductModel(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    created_at = Column(DateTime, default=func.now()) # 커서 페이지네이션을 위해 추가됨

Base.metadata.create_all(bind=engine)

class Product(BaseModel):
    id: int
    name: str
    description: str
    created_at: datetime # 응답에 created_at 포함

    class Config:
        orm_mode = True

app = FastAPI()

# (get_db 함수는 동일)

class CursorParams:
    def __init__(
        self,
        limit: int = Query(10, ge=1, le=100),
        after_cursor: Optional[str] = Query(None, description="다음 페이지의 커서"),
    ):
        self.limit = limit
        self.after_cursor = after_cursor

def decode_cursor(encoded_cursor: str) -> tuple[datetime, int]:
    try:
        decoded_string = base64.b64decode(encoded_cursor).decode('utf-8')
        timestamp_str, item_id_str = decoded_string.split(