Implémenter la pagination

Le pilote mssql-python prend en charge des motifs de pagination efficaces pour diviser de grands ensembles de résultats en segments gérables (pages). Cette approche améliore :

Performance de l’application.
Utilisation de la mémoire.
Expérience utilisateur.
Efficacité réseau.

OFFSET-FETCH pagination

La méthode préférée pour SQL Server 2012 et versions ultérieures :

OFFSET-FETCH de base

Utilisez OFFSET-FETCH pour récupérer une page spécifique de résultats triés :

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}")

Avec le nombre total

Combinez les données de la page avec un total d’enregistrements pour afficher les contrôles de pagination :

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)")

Classe d’aide à la pagination

Créez une classe réutilisable pour gérer les résultats paginés et calculer les propriétés de pagination :

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}")

Pagination par ensemble de clés

Plus efficace pour de grands ensembles de données et une pagination profonde :

Utiliser la pagination par keyset (méthode « seek »)

La pagination par keyset utilise une colonne unique pour accéder à la page suivante, évitant ainsi les scans OFFSET coûteux :

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

Ensemble de clés avec clé composite

Pour les tables avec plusieurs colonnes de tri, utilisez une clé composite pour garantir une pagination stable :

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)

Pagination basée sur le curseur

Encode/décodage des curseurs

Chiffrez l’état de pagination dans une chaîne de curseurs opaque pour les clients 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 pagination

Pour la compatibilité avec les anciennes versions de 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()

Pagination filtrée et triée

Avec des filtres dynamiques

Combinez les clauses WHERE, le tri et OFFSET-FETCH pour prendre en charge les filtres de recherche dynamiques avec pagination. Cet exemple réutilise la PagedResult classe de la classe Pagination helper.

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"
)

Itération basée sur un générateur

Parcourez tous les résultats page par page

Utilisez un générateur pour traiter efficacement tous les enregistrements en fournissant les résultats page par page :

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

Astuces pour les performances

Utilisez des index appropriés

Créez des index sur les colonnes utilisées dans les clauses ORDER BY et WHERE pour optimiser les requêtes de pagination :

-- 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);

Évitez la pagination OFFSET profonde

Les scans OFFSET sautaient des lignes, rendant les pages profondes plus lentes. Utilisez la pagination par keyset pour de meilleures performances :

# 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

Compte total du cache

Évitez de recompter les lignes à chaque demande de pagination en mettant en cache le total pendant une période de temps :

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

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-07-22