0

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

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}%"))

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

“无”类型的表达式不能分配给声明的类型“查询”

果然，filter似乎返回None：

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

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

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，希望得到一些关于这里发生了什么的指导！

4

1 回答 1

2

你错过了两件事：

  • 您需要为 SQLAlchemy 安装打字存根。
  • 该Query.filter()方法有一个装饰器，它定义了返回的内容。

SQLAlchemy 的键入存根

您要安装该sqlalchemy-stubs项目，它为 SQLAlchemy API 提供了存根。

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

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

带有 Query.filter 智能感知信息窗口的 Visual Studio 代码屏幕截图

Query.filter()装饰细节

Query.filter()方法实现实际上并不是对原始实例对象进行操作；它已经用装饰器进行了注释：

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

这@_generative(...)部分在这里很重要；装饰器工厂的定义表明该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返回的是另一个实例。

于 2021-01-18T11:44:27.820 回答