Pony ORM+SQLite：日期转换与any()用法优化技术求助

我正在用Pony ORM开发Web应用教程，现有Python代码能正常运行但实现繁琐，急需简化，问题包括手动转换日期字段、用字符串拼接处理一对多关系。以下是相关代码和Schema，求简化方案：

数据库Schema #

-- staff members: staff_id is primary key
CREATE TABLE staff (
    staff_id BIGINT, 
    personal TEXT, 
    family TEXT
);

-- experiments: sample_id is primary key
CREATE TABLE experiment (
    sample_id BIGINT, 
    kind TEXT, 
    start TEXT, 
    "end" TEXT -- may be NULL if the experiment is ongoing
);

-- join table showing which staff were involved in which experiment
-- there is at least one staff member associated with every experiment
CREATE TABLE performed (
    staff_id BIGINT, 
    sample_id BIGINT
);

-- plate_id is primary key; plate-to-experiment is many to one
-- there is at least one plate associated with each experiment
CREATE TABLE plate (
    plate_id BIGINT, 
    sample_id BIGINT, 
    date TEXT, 
    filename TEXT
);

-- invalidated plates (along with who invalidated the plate and when)
-- this table only contains records for plates that have been invalidated,
-- and contains at most one such record for each plate
CREATE TABLE invalidated (
    plate_id BIGINT, 
    staff_id BIGINT, 
    date TEXT
);

Pony实体类 #

class Staff(DB.Entity):
    staff_id = orm.PrimaryKey(int)
    personal = orm.Required(str)
    family = orm.Required(str)
    performed = orm.Set("Performed")
    invalidated = orm.Set("Invalidated")

class Experiment(DB.Entity):
    sample_id = orm.PrimaryKey(int)
    kind = orm.Required(str)
    start = orm.Required(str)
    end = orm.Optional(str)
    performed = orm.Set("Performed")
    plate = orm.Set("Plate")

class Performed(DB.Entity):
    staff_id = orm.Required(Staff)
    sample_id = orm.Required(Experiment)
    orm.PrimaryKey(staff_id, sample_id)

class Plate(DB.Entity):
    plate_id = orm.PrimaryKey(int)
    sample_id = orm.Required(Experiment)
    date = orm.Required(date)
    filename = orm.Required(str)
    invalidated = orm.Set("Invalidated")

class Invalidated(DB.Entity):
    plate_id = orm.Required(Plate)
    staff_id = orm.Required(Staff)
    date = orm.Required(date)
    orm.PrimaryKey(plate_id, staff_id)

查询需求与现存问题 #

需要为每个实验获取以下信息：

1. 实验ID（用于构建详情页超链接）
2. 开始和结束日期（结束日期可为None表示实验正在进行）
3. 该实验是否存在失效的实验板
4. 参与实验的所有员工ID及姓名

当前实现的痛点：

1. 需手动将日期字段从文本转换为Python date对象，Pony的str2date无法在查询语句中使用；
2. Pony没有类似count()的any()/all()聚合函数，只能通过count(e.plate.invalidated)!=0判断是否存在失效板；
3. 处理多员工关联时，用group_concat()拼接字符串后再拆分转换，操作繁琐，希望有更简便的方式。

当前查询代码 #

def experiment_details():
    query = orm.select(
        (
            e.sample_id,
            e.start,
            e.end,
            orm.count(e.plate.invalidated) != 0,
            orm.group_concat(
                f"<{s.staff_id}|{s.personal}|{s.family}>"
                for s in Staff
                if e.sample_id in s.performed.sample_id.sample_id
            ),
        )
        for e in Experiment
    )
    rows = list(query.order_by(lambda eid, start, end, inv, staff: eid))
    reformatters = (_reformat_as_is, _reformat_date, _reformat_date, _reformat_as_is, _reformat_staff)
    rows = [[f(r) for (f, r) in zip(reformatters, row)] for row in rows]
    return rows


def _reformat_as_is(text):
    """Do not reformat (used for uniform zipping)."""
    return text


def _reformat_date(text):
    """Reformat possibly-empty date."""
    return None if text is None else str2date(text)
        

def _reformat_staff(text):
    """Convert concatenated staff information back to list of tuples."""
    fields = text.lstrip("<").rstrip(">").split(">,<")
    values = [f.split("|") for f in fields]
    result = [(int(v[0]), v[1], v[2]) for v in values]
    return result

简化方案 #

1. 自动处理日期字段转换 #

在Pony实体类中通过converter参数指定日期转换规则，让Pony自动完成数据库文本到Python date对象的转换：

from pony.orm import Optional, Required
from pony.converting import str2date

def date_converter(val):
    return str2date(val) if val is not None else None

class Experiment(DB.Entity):
    sample_id = orm.PrimaryKey(int)
    kind = orm.Required(str)
    start = orm.Required(str, converter=date_converter, sql_type="TEXT")
    end = orm.Optional(str, converter=date_converter, sql_type="TEXT")
    performed = orm.Set("Performed")
    plate = orm.Set("Plate")

class Plate(DB.Entity):
    plate_id = orm.PrimaryKey(int)
    sample_id = orm.Required(Experiment)
    date = orm.Required(str, converter=date_converter, sql_type="TEXT")
    filename = orm.Required(str)
    invalidated = orm.Set("Invalidated")

class Invalidated(DB.Entity):
    plate_id = orm.Required(Plate)
    staff_id = orm.Required(Staff)
    date = orm.Required(str, converter=date_converter, sql_type="TEXT")
    orm.PrimaryKey(plate_id, staff_id)

配置后查询返回的日期字段直接是Python date对象，无需手动调用格式化函数。

2. 高效判断失效板存在性 #

用Pony的exists()函数替代count()判断，生成更高效的EXISTS SQL语句：

orm.exists(p.invalidated for p in e.plate)

该表达式直接检查是否存在关联的失效板记录，性能优于计数后判断非零。

3. 直接获取员工列表，避免字符串拼接拆分 #

利用Pony的关系导航特性，直接从实验对象获取关联员工，无需字符串拼接拆分：

list((s.staff_id, s.personal, s.family) for s in e.performed.staff_id)

通过e.performed.staff_id直接遍历关联的Staff对象，构造所需的元组列表，一步到位。

简化后的完整代码 #

def experiment_details():
    query = orm.select(
        (
            e.sample_id,
            e.start,
            e.end,
            orm.exists(p.invalidated for p in e.plate),
            list((s.staff_id, s.personal, s.family) for s in e.performed.staff_id)
        )
        for e in Experiment
    )
    return list(query.order_by(lambda eid, start, end, inv, staff: eid))

内容的提问来源于stack exchange，提问作者Greg Wilson