How to Type Hint SQLAlchemy Join Queries in Python

This article will show you how and why to properly type hint your SQLAlchemy join queries. This topic is obscure and somewhat difficult to find information on, but in practice it's pretty straight forward. Typehinting SQLAlchemy joins boils down to returning the query result as a tuple of ORM models, as shown below.

def some_sqlalchemy_join_query() -> list[tuple[Model1, Model2]]:
    stmt = ... # some SQLAlchemy join on Model1 and Model2
    res = session.execute(stmt)
    return list(res.tuples().all()) # Return tuples

The full code example shown later assumes the reader is familiar with the basics of SQLAlchemy, and wants to implement type hinting in their code. If you need more practice, the SQLAlchemy Unified Tutorial is a great place to get more familiar with the framework and well worth your time.

Benefits of typehinting ORM models in Python

ORM stands for Object Relational Mapping. It is a means to programmatically map the tables in our database to objects in code. When using an ORM in Python, we are able to add typehinting to our objects instead of just passing around plain dictionaries. This comes with numerous benefits that will speed your development and ultimately make your code more maintainable and prevent bugs. Properly typehinted Python code will give you autocompletion of fields, static analysis and linting, clean code, and potentially night vision.

Reject dictionaries, embrace objects

Using typehinting on your join queries will make for a much faster and powerful development experience than just converting everything to dictionaries. Traversing dictionaries is fraught with peril, as you may receive unexplained KeyErrors caused by hard-to-find typos. On the other hand, your IDE will let you know if you are referencing a field on an object that does not exist, making the debugging and coding process much smoother.

How to typehint a join query

First, we want to define our domain models. These are classes that represent objects in our database. They represent the types of objects that we expect our queries to return. Let's start with two simple models.

from sqlalchemy import (
    ForeignKey,
    String,
    Integer,
    Identity,
)

from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
)

class Base(DeclarativeBase):
    pass

class EmployeeStatus(Base):
    __tablename__ = "lookup_employee_status"

    employee_status_id: Mapped[int] = mapped_column(Integer, Identity(), primary_key=True)
    employee_status: Mapped[str] = mapped_column(String, nullable=True)

class Personnel(Base):
    __tablename__ = "main_personnel"

    personnel_id: Mapped[int] = mapped_column(Integer, Identity(), primary_key=True)
    first_name: Mapped[str] = mapped_column(String, nullable=True)
    last_name: Mapped[str] = mapped_column(String, nullable=True)
    employee_status_id: Mapped[int] = mapped_column(Integer, ForeignKey("lookup_employee_status.employee_status_id"), nullable=True)

We have just defined a very simple domain model representing Personnel in our organization. There is a simple Many to One relation between Personnel and EmployeeStatus. The following typehinted function joins the Personnel table with EmployeeStatus.

import os
from sqlalchemy import create_engine

# Set up database engine
_ENGINE = create_engine(os.environ["DATABASE_URL"])

def qry_personnel_with_status() -> list[tuple[Personnel, EmployeeStatus]]:
    """Get the workforce tracker query result"""
    with Session(_ENGINE) as session:
        stmt = (
            select(
                Personnel, 
                EmployeeStatus
            )
            .join_from(
                Personnel, 
                EmployeeStaus,
                Personnel.employee_status_id == EmployeeStatus.employee_status_id
            )
        )
        res = session.execute(stmt)
        return list(res.tuples().all())

Notice the return annotation of the function defined above. This is telling us that calling this function will return a list of tuples containing a Personnel, and an EmployeeStatus. Thinking in terms of objects instead of tables and rows, this makes perfect sense: every Personnel comes bundled with an EmployeeStatus.

Now, we can enjoy autocomplete and type checking when accessing our data. Here is an example of how you would access the results:

all_personnel = qry_personnel_with_status()
for personnel_tuple in all_personnel:
    # the first item in the tuple is the Personnel object
    print(personnel_tuple[0].first_name)
    # The second item in the tuple is the EmployeeStatus
    print(personnel_tuple[1].employee_status)

I hope this was helpful. Thanks for reading!