在队列中查找位置编号，peewee 产生非法 sqlite3 查询

Question

我正在尝试在队列中查找票的位置编号。查询在 sqlite3 失败，给出的错误非常奇特：sqlite3.OperationalError: near "(": syntax error。我似乎不明白如何使用列引用。

我尝试过分离成更多的子查询，查看了 peewee、sqlite 和 MySQL 中聚合使用、窗口函数等的无数示例。我还没有找到 peewee 中 fn.ROW_NUMBER 的示例。我检查了多个网站上的查询，他们会重复同样的错误。

peewee 生成的查询：

SELECT ROW_NUMBER() OVER (ORDER BY "t1"."ordinal") AS ordinal_index 
FROM "dbticket" AS t1 
WHERE (((("t1"."called_at" IS ?) AND ("t1"."abandoned_at" IS ?)) AND ("t1"."removed_at" IS ?)) AND ("ticket_id" = ?))

参数：

[None, None, None, 'red-shibe'

负责这个地狱的代码：

from pathlib import Path
import os
THIS_FOLDER = Path(os.path.dirname(os.path.abspath(__file__)))
db_file = THIS_FOLDER / 'queue.db'
sqlite_db = SqliteDatabase(str(db_file), pragmas=[('journal_mode', 'wal')])



class DBTicket(BaseModel):
    ordinal = PrimaryKeyAutoIncrementField()
    ticket_id = TextField(unique=True)
    queue_name = TextField()

    initiated_at = DateTimeField(default=datetime.utcnow)
    called_at = DateTimeField(null=True)
    abandoned_at = DateTimeField(null=True)
    removed_at = DateTimeField(null=True)

    def __str__(self):
        return f'{self.ordinal}:{self.ticket_id}'

in_queue_predicate = (DBTicket.called_at.is_null() &
                      DBTicket.abandoned_at.is_null() &
                      DBTicket.removed_at.is_null())

def ticket_place_number(ticket_id):
    queue = DBTicket.select() \
                    .where(in_queue_predicate)
    place_number = queue.select(fn.ROW_NUMBER().over(order_by=[DBTicket.ordinal]).alias('ordinal_index')) \
                        .where(queue.c.ticket_id == ticket_id) \
                        .scalar()
    return place_number

Answer 1

我没有看到任何直接的问题，除了你可以检查你的 sqlite 版本。窗口函数是相对较新的添加，您需要 3.25.0 或更高版本。

import sqlite3
sqlite3.sqlite_version