How to read rows as model object using SQLAlchemy ORM and cloud spanner

Question

I am writing Restful APIs using FastAPI, SQLAlchemy and Cloud Spanner emulator and I stuck on the most basic operation that is reading a row from the table as a model's class object.

I have created a model which represent a table in database using declarative base.

class Users(Base):
    __tablename__ = "users"
    __table_args__ = (
        Index("email", "email", unique=True),
        Index("mobile_number", "mobile_number", unique=True),
    )

    id = Column(String(36), primary_key=True)
    first_name = Column(String(20), nullable=False)
    last_name = Column(String(20), nullable=False)
    email = Column(String(50), nullable=False)
    mobile_number = Column(String(13), nullable=False)
    password = Column(String(100), nullable=False)
    role = Column(ForeignKey("roles.id"), nullable=False)
    is_active = Column(Boolean, nullable=False, default=False)
    picture = Column(String(100), nullable=True)

    @hybrid_property
    def name(self):
        """To get the full name of the user."""
        return self.first_name + " " + self.last_name

This table has 5 records in it.

I just want to retrieve one based on the provided email so I did this -

users = session.query(Users).filter_by(Users.email=="[email protected]").all()

and I got this error -

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: Query.filter_by() takes 1 positional argument but 2 were given

Then I tried by passing dictionary in filter_by -

users = session.query(Users).filter_by(**{"email":"[email protected]"}).all()

But I got this error -

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2772, in all
    return self._iter().all()
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2915, in _iter
    result = self.session.execute(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1717, in execute
    result = compile_state_cls.orm_setup_cursor_result(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 349, in orm_setup_cursor_result
    return loading.instances(result, querycontext)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 88, in instances
    with util.safe_reraise():
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 210, in raise_
    raise exception
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 69, in instances
    *[
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 70, in <listcomp>
    query_entity.row_processor(context, cursor)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 2631, in row_processor
    _instance = loading._instance_processor(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 715, in _instance_processor
    primary_key_getter = result._tuple_getter(pk_cols)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 961, in _tuple_getter
    return self._metadata._row_as_tuple_getter(keys)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 106, in _row_as_tuple_getter
    indexes = self._indexes_for_keys(keys)
AttributeError: '_NoResultMetaData' object has no attribute '_indexes_for_keys'. Did you mean: '_index_for_key'?

SO again I tried after replacing .all() with .first()

users = session.query(Users).filter_by(**{"email":"[email protected]"}).first()

But I have got no response despite of waiting for 15 mins -

>>> users = session.query(Users).filter_by(**{"email":"[email protected]"}).first()

All I want is to get this row as an object of class Users.

Answer 1

Can you try the following query instead:

users = session.query(Users).filter_by(email=="[email protected]").all()

Answer 2

There are two ways to get users filtered by email in your query:

users = session.query(Users).filter(Users.email == "[email protected]").all()

or

users = session.query(Users).filter_by(email="[email protected]").all()