Query Modeled After Another Successful Query Returns AttributeError: type object 'SqlAlchemyRegistrationVOModel' has no attribute '_query_cls'

Question

RESOLVED by passing a db instance rather than passing the Session class.

---

I am querying my database to find the row where the value in the processed_registration_id column matches the processed_registration_id passed into the method:

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

I get an Attribute Error when I attempt to run the query:

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'

This is the SqlAlchemy Model we have defined:

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}

I have modeled this query after a similar query to one of our other tables; that other query works as expected without returning an 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 is the orm.Session class, you need to pass an instance, that is orm.Session() or the equivalent, depending on how your application creates sessions.