过滤器方法行为异常

Question

我正在尝试将类型提示引入现有代码库，但在尝试键入查询时遇到了问题。

from sqlalchemy.orm.query import Query

class DbContext:
    def __init__(self, db_host, db_port, db_name, db_user, db_password):

        engine = create_engine(...)

        session = sessionmaker(bind=engine)
        self.Session: Session = session(bind=engine)

...

def fetch(context: DbContext, filters: ...):
    sub_query: Query = context.Session.query(...)

在我添加类型提示之前，动态过滤只是一个简单的问题：

if filters.name is not None:
    sub_query = sub_query.filter(
        Person.name.ilike(f"%{filters.name}%"))

但是，现在提示我收到此错误：

类型“None”的表达式不能分配给声明的类型“Query”

果然filter似乎返回None：

（方法）过滤器：（*标准：未知）-> 无

我导航到the source，看来该方法确实没有返回任何内容。

def filter(self, *criterion):
    for criterion in list(criterion):
        criterion = expression._expression_literal_as_text(criterion)

        criterion = self._adapt_clause(criterion, True, True)

        if self._criterion is not None:
            self._criterion = self._criterion & criterion
        else:
            self._criterion = criterion

显然某处存在断开连接，因为将 None 分配给 sub_query 应该会导致提示警告的错误，但我需要执行分配以使过滤真正起作用：

# Does NOT work, filtering is not applied
if filters.name is not None:
  sub_query.filter(
               Person.name.ilike(f"%{filters.name}%"))

# Works but Pylance complains
if filters.name is not None:
  sub_query = sub_query.filter(
               Person.name.ilike(f"%{filters.name}%"))

这是我第一次涉足 Python，希望得到一些关于这里发生了什么的指导！

Answer 1

你错过了两件事：

• 您需要为 SQLAlchemy 安装 typing stubs。
• Query.filter() 方法有一个装饰器，用于定义返回的内容。

为 SQLAlchemy 键入存根

您想安装sqlalchemy-stubs project，它为 SQLAlchemy API 提供了存根。

请注意，即使安装了这个存根，您仍然会看到 Pyright（支持 Pylance 扩展的检查工具）存在问题，因为静态存根不能完全代表 SQLAlchemy API 某些部分的动态特性，例如模型列定义（例如，如果您的 Person 模型有一个名为 name 的列，用 name = Column(String) 定义，那么存根无法告诉 Pyright name 将是一个字符串）。 sqlalchemy-stubs 项目包含一个用于 mypy 类型检查器的插件以更好地处理动态部分，但此类插件不能与其他类型检查器一起使用。

安装存根后，Pylance 可以告诉您filter：

Query.filter()装饰器详情

Query.filter() 方法实现实际上并没有对原始实例对象进行操作；它已用 decorator 注释：

    @_generative(_no_statement_condition, _no_limit_offset)
    def filter(self, *criterion):
        ...

@_generative(...) 部分在这里很重要； definition of the decorator factory 表明 filter() 方法本质上被这个包装方法所取代：

    def generate(fn, *args, **kw):
        self = args[0]._clone()
        for assertion in assertions:
            assertion(self, fn.__name__)
        fn(self, *args[1:], **kw)
        return self

这里，fn 是原始 filter() 方法定义，args[0] 是对 self 的引用，即初始 Query 实例。所以self 被替换为调用self._clone()（基本上，创建一个新实例并复制属性），它运行声明的断言（这里，_no_statement_condition 和_no_limit_offset 是这样的断言），然后运行原始函数在克隆上。

所以，filter() 函数所做的是原地更改克隆实例，因此不必返回任何内容；这由 generate() 包装器处理。正是这种用实用程序包装器替换方法的技巧让 Pyright 误以为返回了 None，但安装了存根后，它知道返回的是另一个 Query 实例。