PythonSQLAlchemySQLitePostgreSQL

Using SQL Functions and More in SQLAlchemy ORM

So far we’ve been querying our database in very simplistic ways, using mostly what the ORM provides, and nothing more. But how do we use the ORM in-order to write some more recognizable queries?

Yonatan Vega24 min read
a wizard writing on a whiteboard
Using an ORM is more than just using the basics that are given to us, the goal is for us to reflect the skills that we have in writing pure SQL queries, in an objectified manner.

Lets begin by creating an engine

As we’ve seen in previous articles, we must first declare a few things

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine('sqlite:///:memory:')
Populating the database

Defining our tables

import datetime
from typing import Optional
from sqlalchemy import (
  String, Text, DateTime, ForeignKey,Numeric, Date
)
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase


class Base(DeclarativeBase):
  pass


class Guild(Base):
  __tablename__ = 'guilds'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
  founded_year: Mapped[Optional[int]]


class Alchemist(Base):
  __tablename__ = 'alchemists'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(100), nullable=False)
  specialization: Mapped[Optional[str]] = mapped_column(String(50))
  joined_date: Mapped[Optional[datetime.date]] = mapped_column(Date)
  origin_id: Mapped[Optional[int]] = mapped_column(ForeignKey("origins.id"), nullable=True)


class Apprentice(Base):
  __tablename__ = 'apprentices'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(100), nullable=False)
  mentor_id: Mapped[int] = mapped_column(ForeignKey("alchemists.id"))
  guild_id: Mapped[Optional[int]] = mapped_column(ForeignKey("guilds.id"))
  start_date: Mapped[datetime.date] = mapped_column(Date)


class Origin(Base):
  __tablename__ = 'origins'
  id: Mapped[int] = mapped_column(primary_key=True)
  region_name: Mapped[str] = mapped_column(String(100), nullable=False)
  description: Mapped[Optional[str]] = mapped_column(Text)


class Potion(Base):
  __tablename__ = 'potions'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(100), nullable=False)
  base_element: Mapped[Optional[str]] = mapped_column(String(50))
  potency: Mapped[Optional[int]]
  creation_cost: Mapped[Optional[Numeric]] = mapped_column(Numeric(10, 2))
  created_by_alchemist_id: Mapped[Optional[int]] = mapped_column(ForeignKey("alchemists.id"))


class Ingredient(Base):
  __tablename__ = 'ingredients'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(100), nullable=False)
  rarity: Mapped[Optional[int]]
  source_origin_id: Mapped[Optional[int]] = mapped_column(ForeignKey("origins.id"))


class Experiment(Base):
  __tablename__ = 'experiments'
  id: Mapped[int] = mapped_column(primary_key=True)
  title: Mapped[str] = mapped_column(String(200), nullable=False)
  alchemist_id: Mapped[int] = mapped_column(ForeignKey("alchemists.id"))
  start_time: Mapped[datetime.datetime] = mapped_column(DateTime, default=datetime.datetime.utcnow)
  duration_hours: Mapped[Optional[float]]
  success: Mapped[Optional[bool]]


class Journal(Base):
  __tablename__ = 'journals'
  id: Mapped[int] = mapped_column(primary_key=True)
  alchemist_id: Mapped[int] = mapped_column(ForeignKey("alchemists.id"))
  entry_date: Mapped[datetime.date] = mapped_column(Date, default=datetime.date.today)
  title: Mapped[str] = mapped_column(String(200))
  entry_text: Mapped[str] = mapped_column(Text)


# Create tables
Base.metadata.create_all(engine)

Populating the database

from sqlalchemy import insert

# 1. Independent tables first
origins_data = [
  {"id": 1, "region_name": "Western Plains", "description": "Vast, arid lands."},
  {"id": 2, "region_name": "Northern Mountains", "description": "Cold peaks, rich in minerals."},
  {"id": 3, "region_name": "Coastal Isles", "description": "Humid islands, unique flora."},
  {"id": 4, "region_name": "Sunken City", "description": "Ancient underwater ruins."},
]

guilds_data = [
  {"id": 101, "name": "The Golden Crucible", "founded_year": 1250},
  {"id": 102, "name": "Order of the Serpent", "founded_year": 980},
  {"id": 103, "name": "Skyfire Artificers", "founded_year": 1600},
]

# 2. Tables dependent on Origins
ingredients_data = [
  {"id": 201, "name": "Quicksilver", "rarity": 7, "source_origin_id": 2},
  {"id": 202, "name": "Dragon Scale", "rarity": 9, "source_origin_id": 2},
  {"id": 203, "name": "Moonpetal Bloom", "rarity": 6, "source_origin_id": 3},
  {"id": 204, "name": "Sunstone Dust", "rarity": 8, "source_origin_id": 1},
  {"id": 205, "name": "Void Salt", "rarity": 10, "source_origin_id": 4},
  {"id": 206, "name": "Iron Bark", "rarity": 3, "source_origin_id": 1},
  {"id": 207, "name": "Glow Worm Fluid", "rarity": 5, "source_origin_id": 3},
]

# 3. Tables dependent on Origins
alchemists_data = [
  {"id": 301, "name": "Elara Vance", "specialization": "Transmutation", "joined_date": datetime.date(1650, 5, 10), "origin_id": 1},
  {"id": 302, "name": "Master Borin", "specialization": "Potions", "joined_date": datetime.date(1632, 8, 21), "origin_id": 2},
  {"id": 303, "name": "Silas Croft", "specialization": "Artifice", "joined_date": datetime.date(1665, 1, 15), "origin_id": 1},
  {"id": 304, "name": "Lysandra", "specialization": None, "joined_date": datetime.date(1610, 11, 30), "origin_id": 3},
  {"id": 305, "name": "Zaltar the Mysterious", "specialization": "Astrology", "joined_date": datetime.date(1400, 1, 1), "origin_id": 4},
{"id": 306, "name": "Boogi The Mischievous", "specialization": "Astrology", "joined_date": datetime.date(1100, 1, 1), "origin_id": None},

]

# 4. Tables dependent on Alchemists, Guilds
apprentices_data = [
  {"id": 401, "name": "Finn", "mentor_id": 301, "guild_id": 101, "start_date": datetime.date(1668, 3, 1)},
  {"id": 402, "name": "Roric", "mentor_id": 302, "guild_id": 102, "start_date": datetime.date(1670, 7, 20)},
  {"id": 403, "name": "Jenna", "mentor_id": 303, "guild_id": 103, "start_date": datetime.date(1671, 1, 5)},
  {"id": 404, "name": "Kael", "mentor_id": 301, "guild_id": 101, "start_date": datetime.date(1672, 9, 12)},
]

potions_data = [
  {"id": 501, "name": "Elixir of Vigor", "base_element": "Fire", "potency": 7, "creation_cost": 50.50, "created_by_alchemist_id": 302},
  {"id": 502, "name": "Draught of Steel Skin", "base_element": "Earth", "potency": 8, "creation_cost": 120.00, "created_by_alchemist_id": 301},
  {"id": 503, "name": "Philter of Insight", "base_element": "Air", "potency": 6, "creation_cost": 75.25, "created_by_alchemist_id": 304},
  {"id": 504, "name": "Tincture of Shadow", "base_element": "Void", "potency": 9, "creation_cost": 250.00, "created_by_alchemist_id": 305},
  {"id": 505, "name": "Restorative Balm", "base_element": "Water", "potency": 5, "creation_cost": 30.00, "created_by_alchemist_id": 302},
]

experiments_data = [
  {"id": 601, "title": "Stabilizing Quicksilver", "alchemist_id": 302, "duration_hours": 5.5, "success": True},
  {"id": 602, "title": "Animating Iron Golem", "alchemist_id": 303, "start_time": datetime.datetime(1670, 4, 10, 8, 0, 0), "duration_hours": 72.0, "success": False},
  {"id": 603, "title": "Lunar Essence Extraction", "alchemist_id": 304, "duration_hours": 8.0, "success": True},
  {"id": 604, "title": "Transmuting Lead to Gold (Attempt 7)", "alchemist_id": 301, "duration_hours": 24.5, "success": False},
]

journals_data = [
  {"id": 701, "alchemist_id": 301, "entry_date": datetime.date(1669, 1, 1), "title": "Year Start Observations", "entry_text": "The resonance chamber requires recalibration..."},
  {"id": 702, "alchemist_id": 302, "entry_date": datetime.date(1669, 1, 5), "title": "Notes on Void Salt", "entry_text": "Highly volatile, requires containment field B."},
  {"id": 703, "alchemist_id": 301, "entry_date": datetime.date(1669, 1, 8), "title": "Failed Gold Transmutation", "entry_text": "Resulted in slag again. Impurities in the lead?"},
  {"id": 704, "alchemist_id": 303, "entry_date": datetime.date(1670, 4, 13), "title": "Golem Autopsy", "entry_text": "Power matrix overload. Rune sequence flawed."},
]

# --- Bulk Insertion using Core insert().values() ---
print("Starting bulk inserts using Core API...")
with Session(engine) as session:
  try:
      # Insert data in dependency order using insert().values()
      print("Inserting Origins...")
      if origins_data: session.execute(insert(Origin).values(origins_data))

      print("Inserting Guilds...")
      if guilds_data: session.execute(insert(Guild).values(guilds_data))

      print("Inserting Ingredients...")
      if ingredients_data: session.execute(insert(Ingredient).values(ingredients_data))

      print("Inserting Alchemists...")
      if alchemists_data: session.execute(insert(Alchemist).values(alchemists_data))

      print("Inserting Apprentices...")
      if apprentices_data: session.execute(insert(Apprentice).values(apprentices_data))

      print("Inserting Potions...")
      if potions_data: session.execute(insert(Potion).values(potions_data))

      print("Inserting Experiments...")
      if experiments_data: session.execute(insert(Experiment).values(experiments_data))

      print("Inserting Journals...")
      if journals_data: session.execute(insert(Journal).values(journals_data))

      # Final commit
      print("Committing all inserts...")
      session.commit()
      print("Bulk inserts committed successfully!")

  except Exception as e:
      print(f"An error occurred during Core bulk insert: {e}")
      session.rollback()
      print("Transaction rolled back.")

Making Selects with Where

Lets start with the basics, selects are pretty straightforward in SQLAlchemy, and look almost identical to their native SQL equivalents

from sqlalchemy.orm import Session
from sqlalchemy import select

with Session(engine) as session:
  stmt_select_where = select(
      Alchemist.name,
      Alchemist.specialization.label('focus') # AS focus
  ).where(
      Alchemist.specialization == 'Transmutation' # WHERE clause
  )
  results = session.execute(stmt_select_where).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} focuses on {alchemist.focus}")

The difference between SQLAlchemy queries and native SQL queries is that the FROM is missing from SQLAlchemy queries, as the SQLAlchemy query constructor deduces the FROM clause automatically by what we provide in the SELECT part of our queries.

SQL equivalent

SELECT
  alchemists.name,
  alchemists.specialization AS focus
FROM alchemists
WHERE alchemists.specialization = 'Transmutation';

The great thing about this syntax is that basically any developer who is familiar with SQL can probably read the code example above, this kind of syntax makes SQLAlchemy great for data science where the same developers doing data analysis can also query the database in a very similar way to how they would the database.

🧙‍♂️ One of my methods of writing ORM calls to my databases is to sometimes begin querying my database directly.

Once my query seems to be efficient enough, I then translate it into code, while attempting to maintain the structure of my queries.

Ordering with order_by

Let just dive into the code as I believe the code basically explains itself in this case

from sqlalchemy import desc, asc

with Session(engine) as session:
  stmt_order_by = select(
      Alchemist.name, Alchemist.joined_date
  ).order_by(
      desc(Alchemist.joined_date) # ORDER BY joined_date DESC
      # asc(Alchemist.joined_date) # for ASC (default)
  )
  results = session.execute(stmt_order_by).all()
  print(f"Alchemists by joined date (newest first): {results}")

One confusing thing I experienced when I just started working with SQLAlchemy is the order_by function, which in most cases would accept either a desc or a asc function call.

You do have to import these functions or you can do the following

with Session(engine) as session:
  stmt_order_by = select(
      Alchemist.name, Alchemist.joined_date
  ).order_by(
      Alchemist.joined_date.desc()
  )
  results = session.execute(stmt_order_by).all()
  print(f"Alchemists by joined date (newest first): {results}")

🧙‍♂️ Note that if you omit either function, and just pass a column (<table>.<column>) to the order_by function, the default behavior would be asc

SQL Equivalent

SELECT
  alchemists.name,
  alchemists.joined_date
FROM alchemists
ORDER BY alchemists.joined_date DESC;

Pagination with limit and offset

Implementation pagination in SQLAlchemy is about as straightforward as it would be with native SQL, in the case that you implement it using limit and offset 

with Session(engine) as session:
  stmt_limit_offset = select(
      Alchemist.name
  ).order_by(
      Alchemist.name
  ).limit(2).offset(1) # Skip 1, take 2
  results = session.execute(stmt_limit_offset).scalars().all()
  print(f"Alchemists page 2 (size 2): {results}")

SQL Equivalent

SELECT
  alchemists.name
FROM alchemists
ORDER BY alchemists.name
LIMIT 2 OFFSET 1;

🧙‍♂️ Advanced alchemy note:

In some cases limit and offset are simply not very efficient, mostly in cases where you need to process thousands of records at a time (the SQLAlchemy documentation refers to a small result set as an average of 10,000 rows).

In most of these cases an SQL cursor might be more efficient, both SQLAlchemy Core and the SQLAlchemy ORM support the “yielding” of result sets using the yield_per execution option (through .execution_options )

Selecting Unique Values with DISTINCT

In some cases you might want to have a list of all of the unique values your tables have in the database, lets say you have alchemists in your database each with a specific or repeated specialization (many alchemists may have the same specialization) and you would like to get a list of all of the specialization that exist in your alchemists table, but you do not want the values to repeat.

from sqlalchemy import distinct

with Session(engine) as session:
  stmt_distinct = select(
      distinct(Alchemist.specialization) # DISTINCT specialization
  ).where(
      Alchemist.specialization.is_not(None)
  )
  results = session.execute(stmt_distinct).scalars().all()
  print(f"Distinct Specializations: {results}")

SQL Equivalent

SELECT DISTINCT
  alchemists.specialization
FROM alchemists
WHERE alchemists.specialization IS NOT NULL;

Note that the distinct function is actually changing the SQL statements that is being sent to the database, while other functions such as the Result.unique() function that we will see in a later article does not change the actual SQL statement but is a utility function that is used on the result set returned from the database.

Joining Tables

Even though we will delve into relationships in later articles, join statements are still incredibly useful, there may even be cases where combining join statements as well as loading relationships is a valid design choice (see joinedload later).

Creating joins is straightforward, we can either import the join function or use the .join function on the FromClause object.

🧙‍♂️ The FromClause is an object that is returned from functions such as select

Inner join

with Session(engine) as session:
  stmt_join = select(
      Alchemist.name,
      Origin.region_name
  ).join(Origin, Alchemist.origin_id == Origin.id)
  results = session.execute(stmt_join).all()
  print(f"Alchemists and their Origins (INNER JOIN): {results}")

Note that by default the .join function generates an INNER JOIN sql statement.

🧙‍♂️ Remember that INNER JOIN statements will return only rows from both tables, where there are matching join conditions on both tables.

They might also be referred to simply as JOIN statements.

from sqlalchemy import join

with Session(engine) as session:
  join_condition = join(
      Alchemist,
      Origin,
      Alchemist.origin_id == Origin.id
  )
	
  stmt_join = select(
      Alchemist.name, Origin.region_name
  ).select_from(join_condition)

  results = session.execute(stmt_join).all()
  print(f"Alchemists and their Origins (INNER JOIN): {results}")

🧙‍♂️ The .select_from function is a utility for separating the join target from the select query.

This supports multiple joins, in cases like these .join(...).join(...) which can be useful when you need dynamic join conditions or when you would like to simplify code a little bit.

Power of Mapped Columns when using Joins

I keep saying how great ORM mapped classes are, and how SQLAlchemy handles much of the logic that we need in our queries behind the scenes, and the .join function is no different, in this case, SQLAlchemy can help us write less code, and maybe even make less mistakes with our onclause parameter of the .join function, by having a mapped class, SQLAlchemy can automatically determine the onclause for us!

Meaning we can rewrite our query like so

with Session(engine) as session:
  stmt_join = select(Alchemist.name, Origin.region_name).join(Origin)
  results = session.execute(stmt_join).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} from region {alchemist.region_name}")

Note that we are omitting the onclause parameter and letting SQLAlchemy handle this for us, this also works with the select_from function

with Session(engine) as session:
  stmt_join = select(
      Alchemist.name,
      Origin.region_name
  ).select_from(
      join(Alchemist, Origin)
  )

  results = session.execute(stmt_join).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} from region {alchemist.region_name}")

SQL Equivalent

SELECT
  alchemists.name,
  origins.region_name
FROM alchemists
INNER JOIN origins ON alchemists.origin_id = origins.id;
-- Join is equivalent
-- JOIN origins ON alchemists.origin_id = origins.id;

Left Join

Left join is very useful in the case that the “left” table holds information that may not have rows that relate to it in the “right” table, but we would still like to find the rows on the “left” table that correspond to our filters.

As with innerjoin there are two ways of creating this query, one option is using a select_from and the other is by calling outerjoin directly.

with Session(engine) as session:
  stmt_left_join = select(
      Alchemist.name,
      Origin.region_name
  ).outerjoin(Origin, Alchemist.origin_id == Origin.id)
  results = session.execute(stmt_left_join).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} from region {alchemist.region_name}")

Using an import

from sqlalchemy import outerjoin

with Session(engine) as session:
  stmt_left_join = select(
      Alchemist.name, Origin.region_name
  ).select_from( # Use select_from for clarity with outerjoin
      outerjoin(Alchemist, Origin, Alchemist.origin_id == Origin.id) # Explicit outer join condition
  )
  results = session.execute(stmt_left_join).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} from region {alchemist.region_name}")

If we look at the difference between the INNER LEFT JOIN examples and these OUTER LEFT JOIN examples, we can see that the last Alchemist from the outerjoin examples do not exist in the innerjoin example.

🧙‍♂️ Note that we can omit the onclause in this case too, but I’ve left it in since this article is meant to show equivalencies between SQL and SQLAlchemy.

SQL Equivalent

SELECT
  alchemists.name,
  origins.region_name
FROM alchemists
LEFT OUTER JOIN origins ON alchemists.origin_id = origins.id;

Functions!

SQLAlchemy fully supports the usage of SQL functions in code, both engine specific functions as well as custom ones that you may have in your database, this allows you to use SQL functions that you might’ve used regularly in your SQL queries as part of your SQLAlchemy integration.

Group by, Count and Having

Lets start with the basics, the COUNT function is extremely common in many applications (although it should be used carefully in large databases and complex queries)

from sqlalchemy import func

with Session(engine) as session:
  stmt_group_by = select(
      Alchemist.specialization.label("name"), # specialization AS name
      func.count(Alchemist.id).label("alchemist_count") # COUNT(id) AS alchemist_count
  ).where(
      Alchemist.specialization.is_not(None)
  ).group_by(
      Alchemist.specialization # GROUP BY specialization
  ).having(
      func.count(Alchemist.id) > 1 # HAVING count > 1
  )
  results = session.execute(stmt_group_by).all()
  for spec in results:
      print(f"Specialization {spec.name} has a count of {spec.alchemist_count} alchemists")

🧙‍♂️ The HAVING keyword is used because of the fact that the WHERE keyword cannot be used with aggregate functions.

The func function

func is a special SQLAlchemy function mainly because it will give you type hinting about functions that are known to SQLAlchemy (such as count) and will attempt to execute sql functions that are unknown to SQLAlchemy in the same way as the others, meaning you can use the same syntax for the count function that you would use for your custom functions.

SQL Equivalent

SELECT
  alchemists.specialization,
  COUNT(alchemists.id) AS alchemist_count
FROM alchemists
WHERE alchemists.specialization IS NOT NULL
GROUP BY alchemists.specialization
HAVING COUNT(alchemists.id) > 1;

Unions

A union combines two SELECT statements into a single result set, in the case that both select statements return columns with the same names and types, the result set will combine them both.

from sqlalchemy import union_all

with Session(engine) as session:
  stmt_alch = select(Alchemist.name.label("entity_name"))
  stmt_appr = select(Apprentice.name.label("entity_name"))
  stmt_union = union_all(stmt_alch, stmt_appr) # Combine results including duplicates
  results = session.execute(stmt_union).scalars().all()
  for name in results:
      print(f"Alchemist / Apprentice Name: {name}")
  # Note: union() (without _all) would remove duplicate names

The union_all function will basically return a concatenation of the result sets in the case that the column names and types match.

In contrast, union will not return duplicate values, the way union works is:

  • It executes the first SELECT statement.
  • It executes the second SELECT statement.
  • It combines the results from both statements.
  • It then compares entire rows within the combined result set. If two rows have the exact same values in all the selected columns, regardless of which original table they came from, UNION considers them duplicates and keeps only one unique instance of that row.

SQL Equivalent

SELECT alchemists.name AS entity_name FROM alchemists
UNION ALL
SELECT apprentices.name AS entity_name FROM apprentices;

CTEs (Common Table Expressions)

A common table expression (CTE) is a named temporary result set that exists within the scope of a single statement and that can be referred to later within that statement, possibly multiple times.

CTEs may improve performance of queries in certain situations, mostly in cases of complex queries where you might need to reuse the results of a subquery.

import datetime

with Session(engine) as session:
  joined_date_start = datetime.date(1600, 1, 1)

  recent_alchemists_cte = select(
      Alchemist.id, Alchemist.name
  ).where(
      Alchemist.joined_date >= joined_date_start
  ).cte("recent_alchemists") # Name the CTE (snake_case)

  # Query from the CTE
  stmt_cte = select(
      recent_alchemists_cte.c.name
  ).order_by(
      recent_alchemists_cte.c.name
  )
  results = session.execute(stmt_cte).scalars().all()
  print(f"Recent Alchemists (joined after {joined_date_start}):")
  for name in results:
      print(f"Alchemist Name: {name}")

🧙‍♂️ Not every database system supports CTEs.

In the case that you are using PostgreSQL and need a MATERIALIZED CTE you can use the .prefix_with("MATERIALIZED") function on the cte function.

SQL Equivalent

WITH recent_alchemists AS (
SELECT alchemists.id AS id, alchemists.name AS name 
FROM alchemists 
WHERE alchemists.joined_date >= '1600-01-01'
)
SELECT recent_alchemists.name 
FROM recent_alchemists ORDER BY recent_alchemists.name

Aggregate Functions: SUM() / AVG() / MIN() / MAX()

Just like the functions before, these are easy to use, and are available to us through the func function from SQLAlchemy

with Session(engine) as session:
  stmt_aggregates = select(
      func.sum(Potion.potency).label("total_potency"),
      func.avg(Potion.potency).label("average_potency"),
      func.min(Potion.potency).label("min_potency"),
      func.max(Potion.potency).label("max_potency")
  )
  result = session.execute(stmt_aggregates).first() # Aggregates usually return one row
  print(f"Potion Potency Stats:")
  print(f"- Total Potency: {result.total_potency}")
  print(f"- Average Potency: {result.average_potency}")
  print(f"- Min Potency: {result.min_potency}")
  print(f"- Max Potency: {result.max_potency}")

SQL Equivalent

SELECT
  SUM(potions.potency) AS total_potency,
  AVG(potions.potency) AS average_potency,
  MIN(potions.potency) AS min_potency,
  MAX(potions.potency) AS max_potency
FROM potions;

String Manipulation with LOWER() / UPPER() / LENGTH() / CONCAT

String manipulation is a common task when working within a database

with Session(engine) as session:
  stmt_string_funcs = select(
      func.lower(Alchemist.name).label("lower_name"),
      func.length(Alchemist.name).label("name_length"),
      (Alchemist.name + ' - ' + Alchemist.specialization).label("name_and_focus") # String concatenation
  ).limit(1)
  result = session.execute(stmt_string_funcs).first()
  print("Lowercase Name:", result.lower_name)
  print("Name Length:", result.name_length)
  print("Name and Specialization:", result.name_and_focus)

SQL Equivalent

SELECT
  LOWER(alchemists.name) AS lower_name,
  LENGTH(alchemists.name) AS name_length,
  (alchemists.name || ' - ' || alchemists.specialization) AS name_and_focus
FROM alchemists
LIMIT 1;

Datetime functions and extractions

Datetime methods are some of the most widely used functions in SQL, they are used with most search systems, and as part of the default information we store whenever we push data into the database.

Let look at a few common examples, as well as how to extract parts of our datetime fields

from sqlalchemy import extract

with Session(engine) as session:
  stmt_datetime = select(
      func.now().label("current_ts"), # Current timestamp with timezone
      func.current_date().label("today"), # Current date
      extract('year', Alchemist.joined_date).label("join_year"), # Extract year part
  ).where(Alchemist.joined_date.is_not(None)).limit(1)
  result = session.execute(stmt_datetime).first()
  print("Current Timestamp:", result.current_ts)
  print("Current Date:", result.today)
  print("Joined Year:", result.join_year)

SQL Equivalent (PostgreSQL Example)

SELECT
  NOW() AS current_ts,
  CURRENT_DATE AS today,
  EXTRACT(YEAR FROM alchemists.joined_date) AS join_year
FROM alchemists
WHERE alchemists.joined_date IS NOT NULL
LIMIT 1;

Cases in SQLAlchemy

SQL provides multiple ways of creating cases within our queries, one of them is the CASE keyword and another is the COALESCE function, SQLAlchemy provides the CASE keyword as an importable function and the COALESCE function is available to us through the func function.

Let take a look at an example of cases and coalesce in SQLAlchemy.

from sqlalchemy import case, func

with Session(engine) as session:
  stmt_conditional = select(
      Alchemist.name,
      case(
          (func.count(Apprentice.id) > 1, "Master"), # IF count > 1 THEN 'Master'
          (func.count(Apprentice.id) > 0, "First timer"), # ELSE IF count > 0 THEN 'First timer'
          else_="Not a Teacher" # ELSE 'Not a teacher'
      ).label("teacher_type"),
      func.coalesce(Alchemist.specialization, "Unknown").label("focus_or_unknown") # COALESCE(specialization, 'Unknown')
  ).outerjoin(
      Apprentice
  ).group_by(
      Alchemist.id
  )
  results = session.execute(stmt_conditional).all()
  for alchemist in results:
      print("---")
      print(f"Alchmist name: {alchemist.name}")
      print(f"Teacher type: {alchemist.teacher_type}")
      print(f"Focus or Unknown: {alchemist.focus_or_unknown}")

In this example, we are using a few of the concepts that we’veseen in this article so far, while introducing a few new things.

  • Case allows you to write as many cases that you would want to query against your database in an if-elseif-else type of syntax, in this case we are using it to return a string from the database in each of the cases.
  • The COALESCE function is basically a function form of the Nullish coalescing operator from other programming languages, it either returns a value if it exists, or a default value.
  • We perform an outerjoin on the Apprentice model since we need to check the number of apprentices each master has using the count() function
  • By performing a group_by on the Alchemist.id column we make sure that our aggregation function (count) does not prevent us from getting all of our Alchemist objects.

The code above represents a good example of crafting SQLAlchemy queries that handle some of the logic that we would in code, remember that these functions can be used for much more complex cases where being close to your data could reduce the amount of logic required by the rest of your program.

SQL Equivalent

SELECT
  alchemists.name,
  CASE
      WHEN COUNT(apprentices.id) > 1 THEN 'Master'
      WHEN COUNT(apprentices.id) > 0 THEN 'First timer'
      ELSE 'Not a Teacher'
  END AS teacher_type,
  COALESCE(alchemists.specialization, 'Unknown') AS focus_or_unknown
FROM
  alchemists
LEFT OUTER JOIN
  apprentices ON alchemists.id = apprentices.mentor_id
GROUP BY
  alchemists.id
ORDER BY
  alchemists.name;

Type Casting columns

Type casting is very useful in type rich databases, it is very useful when we want to process our data in a way that is not possible with the current type of a column or when it makes more sense to do it using a different type (for example if we would want to turn a percentage from an integer to a double between 0 and 1 and do some math in our database).

Casting types can also be useful when working with complex JSON (or JSONB, i.e Binary JSON) data, for example if our JSONB objects hold strings representing DATETIME values and we would like to turn them back into DATETIME for usage with comparison operators or other functions.

from sqlalchemy import cast

with Session(engine) as session:
  stmt_cast = select(
      Alchemist.name,
      cast(Alchemist.joined_date, String).label("joined_date_as_string")
  ).limit(1)
  result = session.execute(stmt_cast).first()
  print("Alchemist Name:", result.name)
  print("Joined Date as String:", result.joined_date_as_string)
  # Note: This is just an example.
  # In practice, casting dates to strings is not common (also not recommended).

What is interesting about the cast function is the fact that the type that are asking SQLAlchemy to convert to is not directly an SQL type, instead its the type mapped to the SQL type that we need, in this example String would be cast to VARCHAR (or NVARCHAR in SQL Server).

🧙‍♂️ In a later article we will take a look at creating our own types for SQLAlchemy, which will work with the cast function above.

SQL Equivalent

SELECT
alchemists.name,
CAST(alchemists.joined_date AS VARCHAR) AS joined_date_as_string 
FROM alchemists
LIMIT 1 OFFSET 0

Logical Operators

Native SQL queries are choke full of logical operators such as AND , OR , NOT and the likes.

By default, if you use multiple .where functions in an SQLAlchemy statement the default behavior is to convert these into an AND statement, the same goes for sending multiple expressions into a .where function as *args

i.e:

where(Model.first_name == 'Bruce', Model.last_name == 'Wayne')

If we want to use an OR operator, we need to import this operator as a function from sqlalchemy , the same goes for the AND operator if we would want to be more explicit in our code and make it more readable (even though we can leverage the default behavior).

🧙‍♂️ Digging deeper

It may not seem obvious at first why these three logical operators need special attention from a code writing point of view, all of these operators are available both in SQL and Python, so why cant we just write our queries in python form, and have these keywords mapped to SQL code?

i.e: where(Model.first_name == 'Bruce' and Model.last_name == 'Wayne')

Well, if you think about it, in SQL each logical operation is usually preceded by an expression, to convert python code to SQL expressions you would need some kind of logical differentiation for these expressions.

SQLAlchemy can deal with expressions that include comparison operators in them because Operator Overloading is a supported programming paradigm in Python (as it is in many other programming languages) but it cannot convert python reserved keywords (such as and , or and not) into SQL directly, due to limitations in python itself.

Using the _and, _or, _not functions

Lets start with the OR operator

from sqlalchemy import or_

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      or_(
          Alchemist.specialization == 'Potions',
          Alchemist.joined_date <= datetime.date(1400, 1, 1)
      )
  )
  results = session.execute(stmt_logical_ops).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} is either a Potions specialist or joined before 1400")

Note that you could also write the same function using another operator, with is the | operator, which is the Bitwise OR operator, note that many of the Bitwise Operators of Python also work with SQLAlchemy expressions and logical operators.

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      (Alchemist.specialization == 'Potions') |
      (Alchemist.joined_date <= datetime.date(1400, 1, 1))
  )
  results = session.execute(stmt_logical_ops).all()
  for alchemist in results:
      print(f"Alchemist {alchemist.name} is either a Potions specialist or joined before 1400")

🧙‍♂️ When using Bitwise operators you have to wrap your expressions with parentheses, to differentiate between the assignment side and the bitwise operator.

SQL Equivalent

SELECT alchemists.name 
FROM alchemists 
WHERE alchemists.specialization = 'Potions'
OR alchemists.joined_date <= '1400-01-01'

Lets take a look at the AND operator

from sqlalchemy import and_

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      and_(
          Alchemist.specialization == 'Astrology',
          Alchemist.joined_date > datetime.date(1100, 1, 1)
      )
  )
  results = session.execute(stmt_logical_ops).scalars().all()
  print(f"Logical Ops results: {results}")

The and_ function is as straightforward as the or_ function, and as we’veseen in the introduction of this section, is actually not mandatory.

You can achieve the same functionality using the following syntax as well

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      Alchemist.specialization == 'Astrology',
      Alchemist.joined_date > datetime.date(1100, 1, 1)
  )
  results = session.execute(stmt_logical_ops).scalars().all()
  print(f"Logical Ops results: {results}")

And this works due to the fact that the where function treats all of the expressions passed to just like the and_ function would.

SQL Equivalent

SELECT alchemists.name 
FROM alchemists 
WHERE alchemists.specialization = 'Astrology'
AND alchemists.joined_date > '1100-01-01'

Negating with not_

Can you find the difference between these two snippets of code?

from sqlalchemy import not_

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      not_(Alchemist.origin_id == 1)
  )
  results = session.execute(stmt_logical_ops).scalars().all()
  print(f"Logical Ops results: {results}")

And

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      Alchemist.origin_id != 1
  )
  results = session.execute(stmt_logical_ops).scalars().all()
  print(f"Logical Ops results: {results}")

Found no difference? that’s right.

Thats because in these basic examples, the not_ function is really not required, but it does become an issue not to have it once you look at some other SQL operators that work in conjunction with the sql NOT keyword.

These operators might include the LIKE and IN keywords, in-fact it is very common to see statements like NOT IN when reading SQL code especially in search systems where it might make more sense to exclude rather than include your filtering criteria.

For example

with Session(engine) as session:
  stmt_logical_ops = select(Alchemist.name).where(
      not_(
          Alchemist.specialization.in_(['Potions', 'Transmutation'])
      )
  )
  results = session.execute(stmt_logical_ops).scalars().all()
  print(f"Logical Ops results: {results}")

SQL Equivalent

SELECT alchemists.name 
FROM alchemists 
WHERE (alchemists.specialization NOT IN ('Potions', 'Transmutation'))

Searching for matching strings with LIKE and ILIKE

The LIKE operator is one of those operators that needs careful handling, without the proper indexing you might lose a lot of performance once querying large database tables.

🧙‍♂️ Since each database engine handles string matching in different ways I will not go into the details of improving performance using the LIKE and ILIKE operators.

Keep in mind that if you ever get into a situation where you need to match strings in a large database table, you should probably consult the documentation, do some research about database extensions, different kinds of indexing and experiment quite a bit.

Let take a look at an example

with Session(engine) as session:
  stmt_like = select(Alchemist.name).where(
      or_(
          Alchemist.name.like('E%'), # Starts with E
          Alchemist.name.ilike('%borin%') # Contains 'borin' (case-insensitive)
      )
  )
  results = session.execute(stmt_like).scalars().all()
  print(f"LIKE/ILIKE Ops results: {results}")

SQL Equivalent

SELECT alchemists.name FROM alchemists
WHERE alchemists.name LIKE 'E%'
 OR alchemists.name ILIKE '%borin%';

Contemplating existence with EXISTS

The exists keyword is used as part of a WHERE clause that will resolve to either true of false based off of a subquery that we perform, it is particularly useful when working with large datasets and can be more favorable than a JOIN or WHERE ... IN due to the fact the the EXISTS keyword only needs to find a single match to resolve wether or not the WHERE clause is valid for a specific row.

An example

from sqlalchemy import literal

with Session(engine) as session:
  subq = select(literal(1)).where(
      Alchemist.origin_id == Origin.id
  ).exists()
  stmt_exists = select(Origin.region_name).where(subq)
  results = session.execute(stmt_exists).scalars().all()
  print(f"EXISTS Ops results (Origins with Alchemists): {results}")

You can also use the exists function instead of chaining the .exists function to your subquery statement like so

from sqlalchemy import exists

with Session(engine) as session:
  subq = select(literal(1)).where(
      Alchemist.origin_id == Origin.id
  )
  stmt_exists = select(Origin.region_name).where(exists(subq))
  results = session.execute(stmt_exists).scalars().all()
  print(f"EXISTS Ops results (Origins with Alchemists): {results}")

SQL Equivalent

SELECT o.region_name
FROM origins o
WHERE EXISTS (
  SELECT 1
  FROM alchemists a
  WHERE a.origin_id = o.id
);

Conclusion

In reality there are many more functions and possibilities in SQLAlchemy, almost every paradigm is covered in this library, and if you look hard enough at the documentation you can probably find a way to translate your SQL queries into SQLAlchemy code, even between different dialects, SQLAlchemy provides support for most of the things you might need.

This article is meant to cover a little bit beyond the basics of your everyday SQL uses, hopefully you now have a clearer understanding of constructing SQLAlchemy queries.

In the next article, I cover the Session class and different ways of creating SQLAlchemy sessions before moving on to SQLAlchemy ORM Relationships.

Author image
I'm Yonatan Vega
Owner & Creator, Fullstack.rocks
I've enjoyed writing code ever since I discovered it was an option. At a young age I started by writing code for MMORPG's private servers, and hosting my own private servers at home.

Today I'm writing and architecting software using many different technologies, and I'm always looking for the next thing to learn.