Skip to content

Instantly share code, notes, and snippets.

@gbiz123
Last active January 8, 2024 23:17
Show Gist options
  • Save gbiz123/5e00917f0d67f4b6a9c8eacf60199aed to your computer and use it in GitHub Desktop.
Save gbiz123/5e00917f0d67f4b6a9c8eacf60199aed to your computer and use it in GitHub Desktop.
How to Properly Type Hint SQLAlchemy Join Queries

How to Type Hint SQLAlchemy Join Queries

This Gist will show you how and why to properly type hint your SQLAlchemy join queries. Typehinting join queries is not well documented or discussed as of yet, but I'm here to change that.

If you are converting your ORM models to dictionaries, you are missing out!

If you aren't typehinting in SQLAlchemy, you're missing out on all the powerful ORM features that will speed your development and help prevent bugs. Properly typehinted Python code will give you specific benefits including:

  • Autocompletion of fields
  • Static analysis and linting
  • Clean code

Using typehinting on your join queries will make for a much faster and powerful development experience than just converting everything to dictionaries.

Let's get into the code.

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!

Gregory Bizup

greg@toughdata.net

https://www.toughdata.net

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment