Introduction to SqlAlchemy queries – part 1

The select() is the base function to query data in SQLAlchemy. The select() function translates SQLAlchemy-style query int SQL SELECT adjusted to the applied back-end database. The query mechanism introduced in SQLAlchemy 2.0 and above is different compared to the earlier framework versions.

Database preparation

In the post, I will show how to query records from a single database table. Let’s start with the creation of a database schema. We create a table users and associate it with the User class. Then, we add a few related information user_id, name and first_seen, which we can interpret as a time of account creation. To our class, we add a method __repr__, which presents this information in a readable way.
Finally, we create our database, which will reside in memory and use the SQLite engine. We create the table using the Model class and define the Session object.

from sqlalchemy import Table, Column, Integer, String, create_engine, MetaData
from sqlalchemy.orm import sessionmaker, Mapped, mapped_column, DeclarativeBase, aliased
from sqlalchemy.sql.expression import select
from typing import Optional

class Model(DeclarativeBase):
    pass

class User(Model):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String())
    first_seen: Mapped[str] = mapped_column(String())

    def __repr__(self):
        return "User(user_id={self.id}, " \
                       "name='{self.name}', " \
                       "first_seen={self.first_seen})".format(self=self)


engine = create_engine('sqlite:///:memory:')
Model.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

Test data

When the database is ready, we add two users: Gregory and Maria. You may notice how SQLAlchemy integrates with the code. We created objects representing the User class and added them to the database. You don’t need to write SQL statements and make cumbersome connections between Python code and the embedded code.

with Session() as session:
    user = User(id=1, name='Gregory', first_seen='2024-12-01')
    session.add(user)
    user = User(id=2, name='Maria', first_seen='2024-12-03')
    session.add(user)
    session.commit()

Querying the data

With some rows in the database, the simplest form of reading them is to use the select() function to create a new Select object, which we then invoke using a Session(). The Select object represents an SQL SELECT statement which we invoke using session.execute().

Query all the data

print("Get all users")
smtp = select(User)
print("SQlAlchemy select translated to SQL\n{}".format(smtp))
with Session() as session:
    users = session.execute(smtp)
    for user in users:
        print(user)

The above fragment of code presents two features of SQLAlchemy.

1. SqlAlchemy creates a sqlalchemy.sql.expression.Select object, which we can further filter
2. SqlAlchemy translates the Select object into an SQL statement compatible with your back-end database, which we specified in create_engine(‘sqlite:///:memory:’). If you specify a different database, the SQL statement may be different, especially in more complex queries.

Query all selected rows with first()

When we use first(), we will get only a single record, no matter how many records a database has. As a result, we get a Row object which contains a single User object. In this situation, we don’t need iterations to get the data.

print("Get the first user")
smtp = select(User)
with Session() as session:
    user = session.execute(smtp).first()
    print(user)

Query selected columns

When we invoke the statement below using column name in the Select, we now receive rows with only User.name column:

print("Get only selected columns")
smtp = select(User.name)
with Session() as session:
    user = session.execute(smtp).first()
    print(user)

Query selected rows with where()

When we want to get records which fulfil certain requirements, we use the where() method of the Select object. We receive rows filtered by the User.id column.

print("Get only selected rows using 'where'")
smtp = select(User.name).where(User.id == 2)
with Session() as session:
    user = session.execute(smtp).first()
    print(user)

Query selected rows with get()

The get() function is a special case of where() and filters the rows by the table’s primary key. In our case, the data is filtered by user_id, which is the primary key of the table User.

with Session() as session:
    user = session.get(User, 2)
    print(user)

The whole code is presented below:

Summary

In this post, I presented the most popular methods to query data in SQLAlchemy 2. The queries generate the Select object, which you can further filter using the above-presented methods. Such an approach allows you to integrate nicely database query statements with your Python code. Please note that in an earlier version of SQLAlchemy, the query mechanism is different. Please refer to this post to see the differences.