基于聚合计算字段条件从仓储选数据的代码复用方案咨询

你遇到的问题是典型的业务逻辑分散在领域模型和数据访问层导致的维护冗余问题，下面是几种能把判断逻辑统一到一处的可行方案：

方案1：使用ORM混合属性（Hybrid Property） #

如果你的项目用的是支持混合属性的ORM（比如SQLAlchemy），可以直接把领域模型的判断逻辑同时映射为实例方法和SQL查询表达式，实现一处定义、两处复用：

from sqlalchemy.ext.hybrid import hybrid_property
from datetime import datetime, timedelta
from sqlalchemy import and_, func
from typing import List

class Element:
    status: str
    updated_at: datetime

    @hybrid_property
    def is_ready(self) -> bool:
        # 实例层面的判断逻辑
        return self.status == 'ready' and self.updated_at < datetime.now() - timedelta(hours=1)

    @is_ready.expression
    def is_ready(cls):
        # 自动转换为SQL的查询表达式
        return and_(
            cls.status == 'ready',
            cls.updated_at < func.now() - func.interval('1 hour')
        )

class ElementRepository:
    def __init__(self, session):
        self.session = session

    def get_ready(self) -> List[Element]:
        # 直接复用混合属性作为查询条件
        return self.session.query(Element).filter(Element.is_ready).all()

当需要修改is_ready的判断逻辑时，只需要更新Element类里的两个方法（可以进一步把核心逻辑抽成私有函数避免重复），仓储层代码完全不需要改动。

方案2：Specification模式 #

如果你的业务规则复杂、需要组合多个判断条件，用Specification模式把查询逻辑封装成独立类，同时支持实例校验和查询转换：

from datetime import datetime, timedelta
from sqlalchemy import and_, func
from typing import List

class Element:
    status: str
    updated_at: datetime

    def is_ready(self) -> bool:
        return ReadyElementSpecification.is_satisfied_by(self)

class ReadyElementSpecification:
    @staticmethod
    def is_satisfied_by(element: Element) -> bool:
        # 统一的判断逻辑
        return element.status == 'ready' and element.updated_at < datetime.now() - timedelta(hours=1)

    @staticmethod
    def to_query_condition():
        # 转换为SQL查询条件
        return and_(
            Element.status == 'ready',
            Element.updated_at < func.now() - func.interval('1 hour')
        )

class ElementRepository:
    def __init__(self, session):
        self.session = session

    def get_ready(self) -> List[Element]:
        return self.session.query(Element).filter(ReadyElementSpecification.to_query_condition()).all()

所有判断逻辑都集中在ReadyElementSpecification类中，无论是领域模型的实例校验，还是仓储层的数据库查询，都依赖这个类的方法。后续逻辑变更只需修改这个类即可。

方案3：提取公共逻辑函数 #

如果不想引入复杂模式，最直接的方式是把核心判断逻辑抽成独立的公共函数，让领域模型和仓储层都调用这个函数：

from datetime import datetime, timedelta
from sqlalchemy import and_, func
from typing import List

def is_element_ready(status: str, updated_at: datetime) -> bool:
    return status == 'ready' and updated_at < datetime.now() - timedelta(hours=1)

def get_ready_element_query_condition():
    # 复用核心逻辑生成SQL条件
    return and_(
        Element.status == 'ready',
        Element.updated_at < func.now() - func.interval('1 hour')
    )

class Element:
    status: str
    updated_at: datetime

    def is_ready(self) -> bool:
        return is_element_ready(self.status, self.updated_at)

class ElementRepository:
    def __init__(self, session):
        self.session = session

    def get_ready(self) -> List[Element]:
        return self.session.query(Element).filter(get_ready_element_query_condition()).all()

这种方式简单直接，适合逻辑不复杂的场景，缺点是SQL条件和Python逻辑需要手动保持一致（如果ORM支持动态生成SQL，可以进一步优化）。

内容的提问来源于stack exchange，提问作者Oleg Khamov