好问题！在Peewee中定制SQL方言的语法（比如SQL Server的TOP/OFFSET-FETCH），现在的最佳方式是通过自定义Database子类并重写相关的SQL编译方法——毕竟旧的query compiler相关接口确实已经被更新了。下面是具体的实现步骤和示例：

1. 核心思路：重写SELECT语句的编译逻辑 #

SQL Server的分页语法分为两种场景：

• 仅需返回前N条数据：用SELECT TOP(N) ...替代LIMIT N
• 分页查询（含偏移量）：用OFFSET X ROWS FETCH NEXT Y ROWS ONLY替代OFFSET X LIMIT Y

Peewee的Database类提供了compile_select方法，负责生成最终的SELECT SQL语句，我们可以重写这个方法来实现语法转换。

2. 自定义SQL Relay SQL Server数据库类 #

假设你已经有基于SQL Relay的基础数据库连接逻辑，我们在此基础上扩展：

from peewee import Database, SQL, Clause
from peewee import compiler


class SQLRelaySQLServerDatabase(Database):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # SQL Server用方括号包裹标识符（如表名、列名）
        self.quote_char = '['
        self.unquote_char = ']'

    def compile_select(self, query):
        # 先调用父类方法生成基础的SELECT语句
        base_sql, params = super().compile_select(query)
        
        offset = query._offset
        limit = query._limit
        
        if offset is not None:
            # 处理带偏移量的分页：添加OFFSET和FETCH子句
            offset_clause = SQL(f'OFFSET {offset} ROWS')
            if limit is not None:
                fetch_clause = SQL(f'FETCH NEXT {limit} ROWS ONLY')
                pagination_clause = Clause(offset_clause, fetch_clause)
            else:
                pagination_clause = offset_clause
            
            # 将分页子句追加到基础SQL末尾
            base_sql = f'{base_sql} {pagination_clause}'
        
        elif limit is not None:
            # 处理仅限制条数的情况：在SELECT后插入TOP(N)
            select_pos = base_sql.upper().find('SELECT') + len('SELECT')
            base_sql = f'{base_sql[:select_pos]} TOP({limit}) {base_sql[select_pos:]}'
        
        return base_sql, params

代码说明： #

• 标识符引用：初始化时设置quote_char和unquote_char，确保Peewee生成的SQL符合SQL Server的标识符规则（用[]包裹）
• compile_select重写：
  • 先获取父类生成的基础SELECT语句
  • 判断是否存在offset或limit参数，分别处理两种语法场景
  • 对于带偏移量的分页，直接追加OFFSET...FETCH子句；对于仅限制条数的情况，在SELECT关键字后插入TOP(N)

3. 测试自定义驱动 #

用实际的模型和查询验证语法转换是否正确：

# 初始化数据库连接（替换为你的SQL Relay连接参数）
db = SQLRelaySQLServerDatabase(
    database='your_database',
    host='sql_relay_host',
    port=9000,  # SQL Relay默认端口
    user='your_user',
    password='your_password'
)

# 定义测试模型
class User(db.Model):
    name = db.CharField()
    email = db.CharField()

    class Meta:
        database = db
        table_name = 'users'

# 测试仅限制1条数据的查询
single_user_query = User.select().limit(1)
print(single_user_query.sql())
# 预期输出: ('SELECT TOP(1) [t1].[id], [t1].[name], [t1].[email] FROM [users] AS [t1]', ())

# 测试分页查询（跳过前10条，取接下来10条）
pagination_query = User.select().offset(10).limit(10).order_by(User.id)
print(pagination_query.sql())
# 预期输出: ('SELECT [t1].[id], [t1].[name], [t1].[email] FROM [users] AS [t1] ORDER BY [t1].[id] OFFSET 10 ROWS FETCH NEXT 10 ROWS ONLY', ())

4. 额外注意事项 #

• ORDER BY强制要求：SQL Server的OFFSET子句必须配合ORDER BY使用，否则会报错。你可以在驱动中添加检查逻辑，或者提醒用户在分页查询时务必指定排序规则。
• 边缘情况处理：比如当limit为None但指定了offset的场景（SQL Server允许，但实际业务中少见），代码中已经做了兼容。
• SQL Relay适配：确保你的SQL Relay客户端能正确解析和执行生成的SQL语句，若有特殊的参数绑定规则，可能需要额外重写execute或_execute方法。

内容的提问来源于stack exchange，提问作者Nicholas Tulach