Making SQLAlchemy Queries
In this article, I'll go over how to make queries using both imperative and declarative approaches in SQLAlchemy. We'll begin with the imperative approach and move on to the declarative approach, exploring the various ways to interact with our database.
Preparation
In the last article we’ve seen how to define and migrate tables in both the imperative and declarative approaches, if you skipped that article I think it’s important to go back and read up about the differences in approaches and why both approaches even exist, as the fact that there are multiple ways of doing the same might be confusing at first.
We will begin be re-creating and migrating the tables we’ve created in the last article, and make sure our code works as part of this article.
from sqlalchemy import create_engine, Table, Column, Integer, String, ForeignKey, MetaData
# Create a metadata object to allow SQLAlchemy
# to keep track of all of the imperative models
metadata_obj = MetaData()
# Create our engine and connect to an in-memory SQLite database.
# Note that echo=True
engine = create_engine('sqlite:///:memory:', echo=True)
alchemists_table = Table(
'alchemists', # The name of the table
metadata_obj, # The metatada_obj defined earlier
Column('id', Integer, primary_key=True), # For the sake of simplicity, we will use integer as an ID
Column('name', String(45)),
Column('nickname', String(100)),
Column('age', Integer),
)
potions_table = Table(
'potions',
metadata_obj,
Column('id', Integer, primary_key=True),
Column('name', String(100)),
Column('effect', String(100)),
Column('duration', Integer),
)
brewed_potions_table = Table(
'brewed_potions',
metadata_obj,
Column('id', Integer, primary_key=True),
Column('potion_id', ForeignKey("potions.id"), nullable=False),
Column('alchemist_id', ForeignKey("alchemists.id"), nullable=False),
)
flasks_table = Table(
"flasks",
metadata_obj,
Column('id', Integer, primary_key=True),
Column('type', String(100)),
Column('volume', Integer),
)
# Create all of the imperative models tracked by SQLAlchemy using the engine object
metadata_obj.create_all(engine)Using echo=True
In some parts of this article series we will be using the echo=True flag when creating our engine, this will allow us to see all of the queries that are being sent to the database, and the results that are being returned.
This is a great way to debug our application, in this article series we will mostly use it for new topics, as I believe it is important to see what happens behind the scenes when writing SQLAlchemy code.
If you would like to turn it off, you can edit the code in the last code block to remove the echo=True flag.
Imperative Queries
Once our tables are declared and migrated, we can perform CRUD operations. Since our tables are declared imperatively, we do not create objects from them but leverage them to build database queries.
Inserting data into our tables
Let's begin by populating our tables with data. For this, we'll start by inserting information using a new approach that hasn't been covered in previous articles:
with engine.begin() as connection:
edward = alchemists_table.insert().values(
name='Edward Elric',
nickname='The Fullmetal Alchemist',
age=16
)
al = alchemists_table.insert().values(
name='Alphonse Elric',
nickname='Al',
age=15
)
roy = alchemists_table.insert().values(
name='Roy Mustang',
nickname='Hero of Ishval',
age=29
)
connection.execute(edward)
connection.execute(al)
connection.execute(roy)We can also condense our inserts by writing something similar:
with engine.begin() as connection:
stmt = alchemists_table.insert().values([
{
'name': 'Izumi Curtis',
'nickname': 'Teacher',
'age': 35
},
{
'name': 'Van Hohenheim',
'nickname': 'The Sage of the West',
'age': 451
}
])
connection.execute(stmt)Note that we are using the begin() function, which will automatically commit our insert statements (stmt) when the context manager exits.
In the previous chapter, we covered the difference between the begin() and connect() functions.
Insert with return
We can insert values and then make a select query to get what we just inserted, but obviously, there is a better way in which we construct a query that inserts data and returns it at the same time. We can do so by:
with engine.begin() as connection:
stmt = flasks_table.insert().values([
{
'type': 'Volumetric Flask',
'volume': 1000
},
{
'type': 'Round Bottom Flask',
'volume': 2000
}
]).returning(
flasks_table.c.id,
flasks_table.c.type,
flasks_table.c.volume
)
result = connection.execute(stmt)
print(result.all())As we can see, since the id column is automatically incremented, we can get it right after the insert using the returning function.
Selecting data
To select our data, we also have a very straightforward method to do so using SQLAlchemy:
from sqlalchemy import select
with engine.connect() as connection:
stmt = select(
alchemists_table.c.name,
alchemists_table.c.nickname
).where(alchemists_table.c.age > 30)
results = connection.execute(stmt)
# Pull all of the results from the CursorResult object
result_set = results.all()
for result in result_set:
# print(result.all())
print(result)In this query, we are using some new functions that were not covered in previous chapters:
selectis a function that returns a select query expression; it expects either columns, or full table objects.whereis a function that exists on theselectfunction that we can use in the construction of our query
To our select statement, we pass the column names we would like to have in our result.
When we query imperatively defined tables, we access the column using the .c.column_name accessor, just as we did in the previous chapter when we inspected our table's columns.
We do the same thing for our where statement, but when we do, we are actually passing a comparison expression to the where function, in this case:
alchemists_table.c.age > 30If you are wondering how it is possible for us to make Pythonic comparisons on column definitions, behind the scenes, SQLAlchemy maps Python types to SQL types, which it represents as different classes, which allows it to define custom comparison operators for usage in the query builder (if you're interested in how that works, I recommend reading about dunder methods, also known as magic methods).
In fact, SQLAlchemy allows us to create our own custom types and comparison operations.
This means that if our program had some complex datatypes for which we would want to create custom comparisons, we could do that in a pretty straightforward way.
Updating data
Updating data is just as simple as selecting and inserting. Let's say we want to update the age of an alchemist:
from sqlalchemy import update
with engine.begin() as connection:
stmt = update(alchemists_table).where(
alchemists_table.c.name == 'Edward Elric'
).values(
age=17
)
result = connection.execute(stmt)
print(f"Updated {result.rowcount} rows")In this query, we're using the update function from SQLAlchemy. We first specify which table we want to update, then we use the where function to specify which rows we want to update, and finally, we use the values function to specify what values we want to set.
We can also update multiple rows at once:
with engine.begin() as connection:
stmt = update(alchemists_table).where(
alchemists_table.c.age < 18
).values(
nickname=alchemists_table.c.nickname + ' (Minor)'
)
result = connection.execute(stmt)
print(f"Updated {result.rowcount} rows")In this example, we're updating the nickname of all alchemists who are minors, appending " (Minor)" to their nickname. Notice how we can use column references in the values function as well.
Deleting data
Deleting data is similar to selecting and updating. Let's say we want to delete an alchemist:
from sqlalchemy import delete
with engine.begin() as connection:
stmt = delete(alchemists_table).where(
alchemists_table.c.name == 'Van Hohenheim'
)
result = connection.execute(stmt)
print(f"Deleted {result.rowcount} rows")In this query, we're using the delete function from SQLAlchemy. We first specify which table we want to delete from, then we use the where function to specify which rows we want to delete.
We can also delete multiple rows at once:
with engine.begin() as connection:
stmt = delete(alchemists_table).where(
alchemists_table.c.age > 400
)
result = connection.execute(stmt)
print(f"Deleted {result.rowcount} rows")In this example, we're deleting all alchemists who are older than 400 years.
It's worth noting that when we delete data from a table, any data that depends on it through foreign key constraints will also be affected.
Depending on how the foreign key constraint is set up, the dependent data might be deleted (CASCADE), set to NULL (SET NULL), or the delete operation might be prevented (RESTRICT).
Joining tables
Now that we have inserted data into our tables, we can start joining them to get more complex results. Let's say we have already created and populated our potions_table and brewed_potions_table from the previous chapter:
from sqlalchemy import join, select
with engine.connect() as connection:
# First, let's create a join object
j = alchemists_table.join(
brewed_potions_table,
alchemists_table.c.id == brewed_potions_table.c.alchemist_id
).join(
potions_table,
brewed_potions_table.c.potion_id == potions_table.c.id
)
# Now, let's use this join in a select statement
stmt = select(
alchemists_table.c.name,
potions_table.c.name.label('potion_name')
).select_from(j)
# Execute the query and get the results
result = connection.execute(stmt)
# Print the results
for row in result:
print(f"{row.name} brewed {row.potion_name}")In this query, we're first creating a join object j that joins the alchemists_table with the brewed_potions_table on the id and alchemist_id columns, and then joins the result with the potions_table on the potion_id and id columns.
Then, we're using this join object in a select statement to get the name of the alchemist and the name of the potion. We're using the label function to give the potion name a different label in the result, so we can access it as row.potion_name.
We can also perform more complex joins, such as outer joins:
from sqlalchemy import select, outerjoin
with engine.connect() as connection:
# Create an outer join
j = alchemists_table.outerjoin(
brewed_potions_table,
alchemists_table.c.id == brewed_potions_table.c.alchemist_id
).outerjoin(
potions_table,
brewed_potions_table.c.potion_id == potions_table.c.id
)
# Use the join in a select statement
stmt = select(
alchemists_table.c.name,
potions_table.c.name.label('potion_name')
).select_from(j)
# Execute and get results
result = connection.execute(stmt)
# Print results
for row in result:
potion_name = row.potion_name if row.potion_name else "no potion"
print(f"{row.name} brewed {potion_name}")In this example, we're using outer joins, which means that we'll get all alchemists, even those who haven't brewed any potions. For those alchemists, the potion_name will be None, which is why we're using a conditional to replace None with "no potion".
Declarative Queries
Now that we've covered the imperative approach, let's move on to the declarative approach.
Declarative queries tend to look a little bit different than their imperative counterparts. They are much more ORM-like, as they use objects and classes much more, when working with the Declarative approach you get the benefits of working with very specific types, most IDE’s and text-editors will know how to assist you when working with this approach, leaving the guessing game out of the development flow.
Defining our tables
In the last chapter we’ve taken a look at how to create declarative type models, lets recreate them and move on to using them.
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from sqlalchemy import String, Integer, ForeignKey
class BaseModel(DeclarativeBase):
pass
class Alchemist(BaseModel):
__tablename__ = 'alchemists'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(45), nullable=False)
nickname: Mapped[str] = mapped_column(String(100))
age: Mapped[int]
class Potion(BaseModel):
__tablename__ = 'potions'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(100))
effect: Mapped[str] = mapped_column(String(100))
duration: Mapped[int] = mapped_column(Integer)
class BrewedPotion(BaseModel):
__tablename__ = 'brewed_potions'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
potion_id: Mapped[int] = mapped_column(ForeignKey("potions.id"), nullable=False)
alchemist_id: Mapped[int] = mapped_column(ForeignKey("alchemists.id"), nullable=False)
BaseModel.metadata.create_all(engine)Creating a Session
To query declarative models, we need to create a Session:
from sqlalchemy.orm import Session
# Create a session
with Session(engine) as session:
# Now we can query our declarative models
passThe Session object is a factory for creating SQL statements and handles object state management.
Inserting Data
To insert data using the declarative approach, we create instances of our models and add them to the session:
from sqlalchemy.orm import Session
with Session(engine) as session:
# Create a new alchemist
edward = Alchemist(name='Edward Elric', nickname='Fullmetal Alchemist', age=16)
# Add it to the session
session.add(edward)
# Create and add multiple alchemists at once
al = Alchemist(name='Alphonse Elric', nickname='Al', age=15)
roy = Alchemist(name='Roy Mustang', nickname='Hero of Ishval', age=29)
van = Alchemist(name='Van Hohenheim', nickname='The Sage of the West', age=451)
session.add_all([al, roy, van])
healing_potion = Potion(name='Healing Potion', effect='Heals 50 HP', duration=0)
strength_potion = Potion(name='Strength Potion', effect='Increases strength by 10', duration=300)
invisibility_potion = Potion(name='Invisibility Potion', effect='Grants invisibility', duration=60)
session.add_all([healing_potion, strength_potion, invisibility_potion])
session.commit()
bp1 = BrewedPotion(alchemist_id=al.id, potion_id=healing_potion.id)
bp2 = BrewedPotion(alchemist_id=roy.id, potion_id=strength_potion.id)
bp3 = BrewedPotion(alchemist_id=van.id, potion_id=invisibility_potion.id)
session.add_all([bp1, bp2, bp3])
session.commit()
# Commit the changes
session.commit()
# Now the alchemists have IDs
print(f"Edward's ID: {edward.id}")In this example, we're creating instances of our Alchemist model and adding them to the session. The add method adds a single instance, while the add_all method adds multiple instances at once. After we commit the session, the instances will have their IDs set.
Selecting Data
To select data using the declarative approach, we can use the query method of the session:
with Session(engine) as session:
# Get all alchemists
alchemists = session.query(Alchemist).all()
for alchemist in alchemists:
print(f"{alchemist.name}, {alchemist.nickname}, {alchemist.age}")
# Get alchemists older than 30
older_alchemists = session.query(Alchemist).filter(Alchemist.age > 30).all()
for alchemist in older_alchemists:
print(f"{alchemist.name} is {alchemist.age} years old")In this example, we're using the query method to get all alchemists, and then to get alchemists older than 30 years old. The all method returns a list of all results.
We can also use the newer style with the select function:
from sqlalchemy import select
with Session(engine) as session:
# Get all alchemists
stmt = select(Alchemist)
alchemists = session.scalars(stmt).all()
for alchemist in alchemists:
print(f"{alchemist.name}, {alchemist.nickname}, {alchemist.age}")
# Get alchemists older than 30
stmt = select(Alchemist).where(Alchemist.age > 30)
older_alchemists = session.scalars(stmt).all()
for alchemist in older_alchemists:
print(f"{alchemist.name} is {alchemist.age} years old")In this newer style, we're using the select function to create a select statement, and then we're using the scalars method of the session to execute the statement and get a list of scalar results.
Updating Data
To update data using the declarative approach, we first select the objects we want to update, change their attributes, and then commit the session:
with Session(engine) as session:
# Get Edward Elric
edward = session.query(Alchemist).filter_by(name='Edward Elric').first()
# Update his age
edward.age = 17
# Commit the changes
session.commit()In this example, we're first getting the alchemist named Edward Elric, then we're updating his age, and finally we're committing the changes to the database.
We can also update multiple objects at once:
with Session(engine) as session:
# Get all minor alchemists
minors = session.query(Alchemist).filter(Alchemist.age < 18).all()
# Update their nicknames
for minor in minors:
minor.nickname = minor.nickname + ' (Minor)'
# Commit the changes
session.commit()In this example, we're getting all alchemists who are minors, updating their nicknames, and committing the changes.
Deleting Data
To delete data using the declarative approach, we first select the objects we want to delete, call the delete method on them, and then commit the session:
with Session(engine) as session:
# Get the alchemist named Van Hohenheim
hohenheim = session.query(Alchemist).filter_by(name='Van Hohenheim').first()
# Delete him
session.delete(hohenheim)
# Commit the changes
session.commit()In this example, we're first getting the alchemist named Van Hohenheim, then we're deleting him, and finally we're committing the changes to the database.
We can also delete multiple objects at once:
with Session(engine) as session:
# Get all alchemists older than 400 years
old_alchemists = session.query(Alchemist).filter(Alchemist.age > 400).all()
# Delete them
for alchemist in old_alchemists:
session.delete(alchemist)
# Commit the changes
session.commit()In this example, we're getting all alchemists who are older than 400 years, deleting them, and committing the changes.
Joining Tables
To join tables using the declarative approach, we can use the join method of the query object:
with Session(engine) as session:
# Query alchemists and their potions
query = session.query(
Alchemist.name,
Potion.name.label('potion_name')
).join(
BrewedPotion,
Alchemist.id == BrewedPotion.alchemist_id
).join(
Potion,
BrewedPotion.potion_id == Potion.id
)
# Execute the query and get the results
results = query.all()
# Print the results
for result in results:
print(f"{result.name} brewed {result.potion_name}")In this example, we're joining the Alchemist, BrewedPotion, and Potion tables to get a list of alchemists and the potions they've brewed.
In the next article, we will delve into relationships, which are another way of joining between tables using our defined models.
Relationships also allow much more complex connections, and access to related models that might not be as straightforward to implement using native SQL queries.
Conclusion
In this article, we've learned how to make queries using both the imperative and declarative approaches in SQLAlchemy. We've covered inserting, selecting, updating, deleting, and joining data. These are the fundamental operations that you'll use in almost any application that interacts with a database.
The imperative approach is more SQL-like and gives you precise control over the queries, while the declarative approach is more Pythonic and lets you work with objects. Both have their strengths and are useful in different scenarios.
Next, we'll explore SQLAlchemy relationships, how to implement and use them, and how they allow us to write much more straightforward ORM queries.
Today I'm writing and architecting software using many different technologies, and I'm always looking for the next thing to learn.