实现分页

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

键集分页

处理大型数据集和深度分页时效率更高：

使用键集（查找方法）

键集分页使用唯一列来定位到下一页，从而避免代价高昂的 OFFSET 扫描：

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

避免使用 OFFSET 进行深分页

OFFSET 扫描会跳过大量行，导致较深的分页变慢。 使用键集分页以提升性能：

# 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

相关内容

• 执行查询
• 获取数据
• 连接池