模仿成功查询编写SQLAlchemy查询时触发AttributeError问题

我需要查询数据库中processed_registration_id列值与方法入参匹配的行，编写的代码如下：

from sqlalchemy.orm import Session
from sqlalchemy import exc
from pydantic import BaseModel
from models.sqlalchemy_models import SqlAlchemyRegistrationVOModel

class RegistrationVORepo(BaseModel):
    # 根据processed_registration_id返回Registration VO ID
    def get_registration_vo_id(self, processed_registration_id:int, db: Session):
        print(f"******* 正在查找processed_registration_id: {processed_registration_id}")
        registration_vo_object = (
            db.query(SqlAlchemyRegistrationVOModel)
            .filter(SqlAlchemyRegistrationVOModel.processed_registration_id == processed_registration_id)
            .first()
        )
        print(f"******* 返回的registration_vo_id: {registration_vo_object.id}")
        return registration_vo_object.id

运行代码时触发AttributeError，报错信息如下：

2024-02-28 15:39:19   File "/app/./queries/processed_registration_vos.py", line 35, in get_registration_vo_id
2024-02-28 15:39:19     db.query(SqlAlchemyRegistrationVOModel)
2024-02-28 15:39:19   File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2166, in query
2024-02-28 15:39:19     return self._query_cls(entities, self, **kwargs)
2024-02-28 15:39:19 AttributeError: type object 'SqlAlchemyRegistrationVOModel' has no attribute '_query_cls'

对应的SqlAlchemy模型定义：

class SqlAlchemyRegistrationVOModel(Base):
    __tablename__ = "registration_vo"

    id = Column(Integer, primary_key=True)
    phone_number = Column(Integer, nullable=False, unique=True)
    processed_registration_id = Column(Integer, nullable=False, unique=True)

    def as_dict(self):
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

这段查询是模仿另一张表的成功查询编写的，成功代码如下：

# 根据路由中传入的get_current_user查询要更新的用户
user = (
    db.query(SqlAlchemyAccountModel)
    .filter(SqlAlchemyAccountModel.id == sql_alchemy_current_user.id)
    .first()
)

问题原因与解决方法 #

报错的核心原因是调用get_registration_vo_id时，传入的db参数是Session类本身，而非Session的实例对象。SqlAlchemy的query()方法必须通过Session实例调用，直接传入类会导致无法找到模型对应的查询类，触发上述错误。

对比成功的查询代码，它使用的是已经实例化的Session对象（比如通过sessionmaker创建的实例），因此能正常执行。

修正方式：确保调用get_registration_vo_id时，传入的是Session实例（例如db = SessionLocal()生成的对象），而不是Session类。

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