페이지 매김 구현

mssql-python 드라이버는 큰 결과 세트를 관리 가능한 청크(페이지)로 나누는 효율적인 페이지네이션 패턴을 지원합니다. 이 접근법은 다음을 개선합니다:

• 애플리케이션 성능.
• 메모리 사용량.
• 사용자 환경.
• 네트워크 효율성.

OFFSET-FETCH 페이지 매김

SQL Server 2012 및 이후 버전에서 선호되는 방법:

기본 OFFSET-FETCH

정렬된 결과의 특정 페이지를 검색하려면 OFFSET-FETCH를 사용합니다:

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

def get_page(cursor, page: int, page_size: int) -> list:
    """Get a specific page of results."""
    offset = (page - 1) * page_size
    
    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY Name
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})
    
    return cursor.fetchall()

# Get page 3 with 20 items per page
products = get_page(cursor, page=3, page_size=20)
for p in products:
    print(f"{p.ProductID}: {p.Name}")

총 수와 함께

페이지 데이터를 전체 레코드 수와 결합하여 페이지네이션 제어를 표시하세요:

def get_page_with_count(cursor, page: int, page_size: int) -> tuple[list, int]:
    """Get page results and total count."""
    offset = (page - 1) * page_size
    
    # Get total count
    cursor.execute("SELECT COUNT(*) FROM Production.Product")
    total_count = cursor.fetchval()
    
    # Get page data
    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY Name
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})
    
    return cursor.fetchall(), total_count

products, total = get_page_with_count(cursor, page=1, page_size=20)
total_pages = (total + 19) // 20  # Ceiling division
print(f"Page 1 of {total_pages} ({total} total products)")

페이지네이션 헬퍼 클래스

페이지 결과를 관리하고 페이지네이션 속성을 계산하기 위해 재사용 가능한 클래스를 생성합니다:

from dataclasses import dataclass
from typing import Generic, TypeVar, List

T = TypeVar('T')

@dataclass
class PagedResult(Generic[T]):
    """Container for paged query results."""
    items: List[T]
    page: int
    page_size: int
    total_count: int
    
    @property
    def total_pages(self) -> int:
        return (self.total_count + self.page_size - 1) // self.page_size
    
    @property
    def has_previous(self) -> bool:
        return self.page > 1
    
    @property
    def has_next(self) -> bool:
        return self.page < self.total_pages

def get_products_paged(cursor, page: int, page_size: int = 20) -> PagedResult:
    """Get paged product results."""
    offset = (page - 1) * page_size
    
    # Single query with COUNT OVER for total
    cursor.execute("""
        SELECT 
            ProductID, Name, ListPrice,
            COUNT(*) OVER() AS TotalCount
        FROM Production.Product
        ORDER BY Name
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})
    
    rows = cursor.fetchall()
    total = rows[0].TotalCount if rows else 0
    
    return PagedResult(
        items=rows,
        page=page,
        page_size=page_size,
        total_count=total
    )

# Usage
result = get_products_paged(cursor, page=2, page_size=10)
print(f"Page {result.page} of {result.total_pages}")
print(f"Has previous: {result.has_previous}, Has next: {result.has_next}")

키셋 페이지네이션

대규모 데이터셋과 깊은 페이징에 더 효율적인 방법:

키셋 사용(시크 메서드)

키셋 페이지네이션은 고유 열을 사용하여 다음 페이지로 탐색하여 비용이 많이 드는 오프셋 스캔을 피합니다:

def get_products_after(cursor, last_id: int | None, page_size: int = 20) -> list:
    """Get products after a specific ID."""
    if last_id is None:
        # First page
        cursor.execute("""
            SELECT TOP (%(page_size)s) ProductID, Name, ListPrice
            FROM Production.Product
            ORDER BY ProductID
        """, {"page_size": page_size})
    else:
        # Subsequent pages
        cursor.execute("""
            SELECT TOP (%(page_size)s) ProductID, Name, ListPrice
            FROM Production.Product
            WHERE ProductID > %(last_id)s
            ORDER BY ProductID
        """, {"page_size": page_size, "last_id": last_id})
    
    return cursor.fetchall()

# Iterate through all products in pages
last_id = None
while True:
    products = get_products_after(cursor, last_id, page_size=100)
    if not products:
        break
    
    for p in products:
        print(f"{p.ProductID}: {p.Name}")
    
    last_id = products[-1].ProductID  # Track last ID for next iteration

컴포지트 키가 포함된 키셋

정렬 열이 여러 개의 테이블에 있을 경우, 안정적인 페이지닝을 위해 복합 키를 사용하세요:

def get_orders_page(cursor, last_date: datetime | None, last_id: int | None, 
                    page_size: int = 20) -> list:
    """Keyset pagination with composite key (OrderDate, SalesOrderID)."""
    if last_date is None:
        cursor.execute("""
            SELECT TOP (%(size)s) SalesOrderID, OrderDate, CustomerID, TotalDue
            FROM Sales.SalesOrderHeader
            ORDER BY OrderDate DESC, SalesOrderID DESC
        """, {"size": page_size})
    else:
        cursor.execute("""
            SELECT TOP (%(size)s) SalesOrderID, OrderDate, CustomerID, TotalDue
            FROM Sales.SalesOrderHeader
            WHERE (OrderDate < %(last_date)s) 
               OR (OrderDate = %(last_date)s AND SalesOrderID < %(last_id)s)
            ORDER BY OrderDate DESC, SalesOrderID DESC
        """, {"size": page_size, "last_date": last_date, "last_id": last_id})
    
    return cursor.fetchall()

# Usage
orders = get_orders_page(cursor, None, None, page_size=50)
if orders:
    # Get cursor for next page
    last = orders[-1]
    next_page = get_orders_page(cursor, last.OrderDate, last.SalesOrderID, page_size=50)

커서 기반 페이지네이션

인코딩/디코딩 커서

API 클라이언트를 위해 페이지네이션 상태를 불투명 커서 문자열로 암호화하세요:

import base64
import json

def encode_cursor(values: dict) -> str:
    """Encode pagination cursor."""
    return base64.b64encode(json.dumps(values).encode()).decode()

def decode_cursor(cursor_str: str) -> dict:
    """Decode pagination cursor."""
    return json.loads(base64.b64decode(cursor_str.encode()).decode())

def get_page_with_cursor(cursor, after: str | None, page_size: int = 20) -> tuple[list, str | None]:
    """Get page using cursor-based pagination."""
    if after is None:
        cursor.execute("""
            SELECT TOP (%(size)s) ProductID, Name, ListPrice
            FROM Production.Product
            ORDER BY ProductID
        """, {"size": page_size})
    else:
        values = decode_cursor(after)
        cursor.execute("""
            SELECT TOP (%(size)s) ProductID, Name, ListPrice
            FROM Production.Product
            WHERE ProductID > %(last_id)s
            ORDER BY ProductID
        """, {"size": page_size, "last_id": values["id"]})
    
    rows = cursor.fetchall()
    
    next_cursor = None
    if rows:
        next_cursor = encode_cursor({"id": rows[-1].ProductID})
    
    return rows, next_cursor

# Usage - First page
products, next_cursor = get_page_with_cursor(cursor, after=None, page_size=20)

# Next page
if next_cursor:
    products, next_cursor = get_page_with_cursor(cursor, after=next_cursor, page_size=20)

ROW_NUMBER 페이지 매김

구버전 SQL Server와의 호환성을 위해:

def get_page_row_number(cursor, page: int, page_size: int) -> list:
    """Pagination using ROW_NUMBER() - works on SQL Server 2005+."""
    cursor.execute("""
        WITH NumberedProducts AS (
            SELECT 
                ProductID, Name, ListPrice,
                ROW_NUMBER() OVER (ORDER BY Name) AS RowNum
            FROM Production.Product
        )
        SELECT ProductID, Name, ListPrice
        FROM NumberedProducts
        WHERE RowNum > %(start)s AND RowNum <= %(end)s
    """, {"start": (page - 1) * page_size, "end": page * page_size})
    
    return cursor.fetchall()

필터링 및 정렬된 페이지네이션

동적 필터와 함께

WHERE 조건 절, 정렬 및 OFFSET-FETCH 페이지 매김이 있는 동적 검색 필터를 지원하도록 결합하세요. 이 예시는 PagedResult의 클래스를 재사용합니다.

def search_products_paged(cursor, search: str | None, category: int | None,
                         page: int = 1, page_size: int = 20, 
                         sort: str = "Name") -> PagedResult:
    """Search products with filters, sorting, and pagination."""
    
    # Build WHERE clause
    conditions = []
    params = {"offset": (page - 1) * page_size, "page_size": page_size}
    
    if search:
        conditions.append("Name LIKE %(search)s")
        params["search"] = f"%{search}%"
    
    if category:
        conditions.append("ProductSubcategoryID = %(category)s")
        params["category"] = category
    
    where_clause = "WHERE " + " AND ".join(conditions) if conditions else ""
    
    # Validate sort column
    sort_columns = {"Name": "Name", "Price": "ListPrice", "ID": "ProductID"}
    order_column = sort_columns.get(sort, "Name")
    
    # Execute query
    query = f"""
        SELECT 
            ProductID, Name, ListPrice, ProductSubcategoryID,
            COUNT(*) OVER() AS TotalCount
        FROM Production.Product
        {where_clause}
        ORDER BY {order_column}
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """
    
    cursor.execute(query, params)
    rows = cursor.fetchall()
    
    return PagedResult(
        items=rows,
        page=page,
        page_size=page_size,
        total_count=rows[0].TotalCount if rows else 0
    )

# Usage
results = search_products_paged(
    cursor, 
    search="Road", 
    category=2,
    page=1, 
    page_size=15,
    sort="Price"
)

생성기 기반 반복

모든 결과를 페이지에 반복하세요

생성기를 사용해 한 페이지씩 결과를 효율적으로 처리하세요:

def iter_all_products(cursor, page_size: int = 1000):
    """Generator that yields all products in pages."""
    offset = 0
    
    while True:
        cursor.execute("""
            SELECT ProductID, Name, ListPrice
            FROM Production.Product
            ORDER BY ProductID
            OFFSET %(offset)s ROWS
            FETCH NEXT %(size)s ROWS ONLY
        """, {"offset": offset, "size": page_size})
        
        rows = cursor.fetchall()
        if not rows:
            break
        
        for row in rows:
            yield row
        
        offset += page_size

# Process all products without loading into memory
for product in iter_all_products(cursor, page_size=500):
    print(product)  # Replace with your own row-handling logic

성능 팁

적절한 색인을 사용하세요

ORDER BY와 WHERE 절에서 사용되는 열에 인덱스를 생성하여 페이지네이션 쿼리를 최적화합니다:

-- Index for OFFSET-FETCH on Name
CREATE INDEX IX_Products_Name ON Production.Product (Name);

-- Index for keyset pagination on ID
CREATE INDEX IX_Products_ID ON Production.Product (ProductID);

-- Covering index for common query
CREATE INDEX IX_Products_SubCategory_Name 
ON Production.Product (ProductSubcategoryID, Name) 
INCLUDE (ListPrice);

깊은 오프셋 페이지네이션을 피하세요

오프셋 스캔이 행을 건너뛰어 깊은 페이지가 느려졌습니다. 더 나은 성능을 위해 키셋 페이지네이션을 사용하세요:

# Slow for deep pages
get_page(cursor, page=1000, page_size=20)  # Scans 19,980 rows first

# Use keyset for better deep-page performance
get_products_after(cursor, last_id=19980, page_size=20)  # Seeks directly

캐시 총 수

일정 기간 동안 총액을 캐싱하여 매번 페이지 요청마다 행을 다시 집계하지 않도록 하세요:

import time

class PaginatedQuery:
    def __init__(self, cursor, count_cache_seconds: int = 60):
        self.cursor = cursor
        self.count_cache_seconds = count_cache_seconds
        self._count_cache = {}
    
    def get_count(self, query_key: str, count_query: str, params: dict) -> int:
        """Get cached count or execute count query."""
        now = time.time()
        
        if query_key in self._count_cache:
            count, timestamp = self._count_cache[query_key]
            if now - timestamp < self.count_cache_seconds:
                return count
        
        self.cursor.execute(count_query, params)
        count = self.cursor.fetchval()
        self._count_cache[query_key] = (count, now)
        return count

관련 콘텐츠

• 쿼리 실행
• 데이터 검색
• 연결 풀링