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.
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. 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.
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. 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. 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. 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: I hope this was helpful.
Thanks for reading!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
Benefits of typehinting ORM models in Python
Reject dictionaries, embrace objects
How to typehint a join query
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)
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())
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)