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 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.
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