另一个成功查询后建模的查询返回 AttributeError：类型对象“SqlAlchemyRegistrationVOModel”没有属性“_query_cls”

Question

我正在查询数据库以查找processed_registration_id列中的值与传递到方法中的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):
    # Return the Registration VO ID for a processed_registration_id.
    def get_registration_vo_id(self, processed_registration_id:int, db: Session):
        print(f"******* Searching for processed_registration id: {processed_registration_id}")
        registration_vo_object = (
            db.query(SqlAlchemyRegistrationVOModel)
            .filter(SqlAlchemyRegistrationVOModel.processed_registration_id == processed_registration_id)
            .first()
        )
        print(f"******* Returned registration_vo_id: {registration_vo_object.id}")
        return registration_vo_object.id

当我尝试运行查询时出现属性错误：

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}

我根据对我们其他表之一的类似查询对该查询进行了建模；其他查询按预期工作而不返回 AttributeError:

        # Queries the database to find the user to update based off of get_current_user passed in router
        user = (
            db.query(SqlAlchemyAccountModel)
            .filter(SqlAlchemyAccountModel.id == sql_alchemy_current_user.id)
            .first()
        )

Answer 1

db

是

orm.Session

类，您需要传递一个实例，即

orm.Session()

或等效项，具体取决于您的应用程序如何创建会话。