SQLAlchemy 对强制映射属性类的过滤器输入支持

问题描述 投票：0回答：1

对于我的应用程序，我试图将我的“域层”类与数据库后端分开，以便能够独立于数据库对这些类进行单元测试。我使用 mypy 进行静态分析，并且在将 

Query.filter()
与我的命令式映射类的属性一起使用时出现输入错误。我不想在所有使用 
# type: ignore
的地方都使用 
Query.filter()
，所以我正在寻找解决此问题的正确方法。

我创建了以下模型：

# project/models/user.py


from attrs import define, field


@define(slots=False, kw_only=True)
class User:
    id: int = field(init=False)
    username: str = field()
    hashed_password: str = field()
    is_active: bool = field(default=True)
    is_superuser: bool = field(default=False)

在生产中，我当然使用具有下表的数据库：

# project/db/tables/user.py

import sqlalchemy as sa
from sqlalchemy.sql import expression

from ..base import mapper_registry


user = sa.Table(
    "users",
    mapper_registry.metadata,
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("username", sa.String(155), unique=True, nullable=False),
    sa.Column("hashed_password", sa.String(200), nullable=False),
    sa.Column("is_active", sa.Boolean, nullable=False, server_default=expression.true()),
    sa.Column(
        "is_superuser", sa.Boolean, nullable=False, server_default=expression.false()
    ),
)

为了将我的类映射到表，我执行了以下操作：

# project/db/__init__.py

from .base import mapper_registry
from .tables import user

from project.models import User


mapper_registry.map_imperatively(User, user)

该功能按预期工作，正在构建正确的查询，但是当我尝试使用模型类的属性调用 

Query.filter()
时，我从 mypy 中收到输入错误，因为 
Query.filter()
需要列表达式而不是普通的 python 类型（我使用与 sqlalchemy 捆绑在一起的 mypy 
sqlalchemy = {version = "^2.0.15", extras = ["mypy"]}
，并将 mypy 插件包含在 mypy.ini 中）

# project/queries.py

from typing import List

from sqlalchemy import true
from sqlalchemy.orm import Session

from project.models import User


def list_active(session: Session) -> List[User]:
    return session.query(User).filter(User.is_active == true()).all()

错误：

project/queries.py12: error: Argument 1 to "filter" of "Query" has incompatible type "bool"; expected "Union[ColumnElement[bool], _HasClauseElement, SQLCoreOperations[bool], ExpressionElementRole[bool], Callable[[], ColumnElement[bool]], LambdaElement]"  [arg-type]
python sqlalchemy mypy python-attrs
1个回答
0
投票

TL;DR： 在我看来，没有“正确”的解决方案，因为文档中没有提到它。

session.query(User).filter(users_table.columns["is_active"].is_(True)).all()

另外，我也不认为命令式映射发挥作用，故事与声明式映射相同（我在我的项目中使用这种风格，并且看到了相同的问题）。

说明：

SQLAlchemy 不适合 mypy 的这种类型检查，据我所知，没有干净的解决方法。不过，这里有一段代码来强调一些想法和“替代方案”：

from typing import List
import sqlalchemy as sa
from sqlalchemy.orm import Session, registry
from sqlalchemy.sql.expression import BinaryExpression
from attrs import define, field

from sqlalchemy.sql import expression

engine = sa.create_engine("sqlite:///test.sqlite3")
mapper_registry = registry()


# Data model and table objects
@define(slots=False, kw_only=True)
class User:
    id: int = field(init=False)
    username: str = field()
    hashed_password: str = field()
    is_active: bool = field(default=True)
    is_superuser: bool = field(default=False)


user = sa.Table(
    "users",
    mapper_registry.metadata,
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("username", sa.String(155), unique=True, nullable=False),
    sa.Column("hashed_password", sa.String(200), nullable=False),
    sa.Column(
        "is_active", sa.Boolean, nullable=False, server_default=expression.true()
    ),
    sa.Column(
        "is_superuser", sa.Boolean, nullable=False, server_default=expression.false()
    ),
)


# Map the User class to the 'users' table imperatively
mapper_registry.map_imperatively(User, user)


# Function to populate the database with initial data
def populate_database():
    # 'users' table creation
    user.create(engine)

    with Session(engine) as session:
        user_1 = User(username="John Doe", hashed_password="aaa")
        user_2 = User(username="Jane Smith", hashed_password="bbb")
        user_3 = User(username="Bob Johnson", hashed_password="ccc")

        session.add(user_1)
        session.add(user_2)
        session.add(user_3)
        session.commit()


def list_active() -> List[User]:
    with Session(engine) as session:
        expr: bool = User.is_active == sa.true()
        result = session.query(User).filter(expr).all()
    return result


def list_active_with_idiom() -> List[User]:
    with Session(engine) as session:
        expr: BinaryExpression = User.is_active.is_(True)
        result = session.query(User).filter(expr).all()
    return result


def list_active_with_types() -> List[User]:
    with Session(engine) as session:
        expr: BinaryExpression = user.columns["is_active"].is_(True)
        result = session.query(User).filter(user.columns["is_active"].is_(True)).all()
    return result


# Run this once only, to populate the table with initial data using the models
# populate_database()

# Run after populating the database
print(list_active())  # question way
print(list_active_with_idiom())  # SQLAlchemy recommended way
print(list_active_with_types())  # type checking sound

该代码片段使用示例的原始数据模型和 Table 对象，只需将其放在同一个文件中即可重现结果。通过调用 

populate_database()
初始化 SQLite DB 后，您可以运行 3 个替代函数来列出活动用户。 
list_active()
是你的原始函数，
list_active_with_idiom()
是使用SQLAlchemy推荐方式的函数，最后一个是类型检查声音函数。

  • list_active()
    计算 Python 表达式并创建一个 
    bool
    对象，该对象被输入到 
    filter()
    函数中，该对象仅针对 
    _ColumnExpressionArgument[bool]
    进行注释（mypy 错误消息中显示的 type）。这是原版。

  • list_active_with_idiom()
    使用 SQLalchemy 方式，如文档中所示。它从 BinaryExpression
     对象、
    Column     关键字和 
    True
    运算符创建所谓的 
    is_()
    （您可以看到它是出于输入和解释目的而导入的）。它们分别是 
    left
     的 
    right
    、
    operator
     和 
    BinaryExpression
     属性。该类的基类是
    OperatorExpression
    ，其中以
    ColumnElement
    为基类（使类型检查通过）。不过，您可以注意到，该行仍然会生成类型检查问题，因为它依赖于从数据模型属性（布尔值）创建此 
    is_()
     运算符。 
    BinaryExpression
     之所以有效，是因为数据模型和表之间存在映射，但 
    User.is_active.is_(True)
     仍然只是 
    mypy
     的布尔值。


  • User.is_active
     是“类型检查声音”版本，并利用您可以从第二个选项中的行为中学到的知识。 
    list_active_with_types()
     首先直接从原始 
    users_table.columns["is_active"].is_(True)
     对象中获取 Column 对象，然后应用相同的运算符与 
    Table
     进行比较。我不确定这就是它应该如何工作，但让 
    True
     高兴。
    

    

    • 如果运行此示例，您应该会看到以下 2 个 
mypy
 错误：
mypy


如您所见，最后一个选项没有类型检查错误。不确定这是否完全正确，但我在官方文档中没有找到它，我将其更多地视为 SQLAlchemy 注释的解决方法。


额外评论：
 像 SQLModel 这样的项目使用 SQLAlchemy 作为 ORM 部分；除其他外，该项目添加了大量注释。阅读源代码后，看起来他们跳过了 mypy 类型检查，以检查一些像这样 one 这样棘手的行。
    


      

      
    

  

  

    
最新问题


  





  

    © www.soinside.com 2019 - 2024. All rights reserved.