Migrating to SQLAlchemy

I have always been a fan of the agile methodology. Sketch, implement, validate, iterate, extend. This has always proven to me that it is the best way to deliver value to stakeholders. It also depends on what the specific needs of a client are, but, usually, they are always happy to see the development of the product and to notice how its value increases as new updates are pushed (users are happy to update, contrary to what one would expect).

Updating a software product can bring new features (fun!) or bug fixes or performance upgrades (it feels faster!). This is one of the main characteristics of software that cannot be compared to other disciplines. Imagine having the architect and the builders build just one floor of the new skyscraper of your office, and they let you use that floor, but the immediate upper floor does not even have a ceiling. Software, due to its nature, allows developers to keep an increasing delivery of value to the stakeholders, which has a positive effect on the user’s perception of the software.

This way of working has always given me good results, as teams in which I was deployed celebrated the fact that we built stuff in different stages (we were lowering the scope and determining how closed-world our scope was, making it easier to handle), and always being in close contact with the customer also helped to better polish and adjust requirements. Of course, in order to avoid refactoring the whole system after each sprint, a good understanding of the problem and how to decompose it is required: a good designed software is able to keep its business logic in one place, and lets its further features to be built on top of the core logic. This avoids unneeded refactors, since the details that need adjustment are usually placed in the most external modules.

Now, for one of my experiments, I have been generating lots of data about a trendy topic, which is typically seasonal, and appears mostly during the summer season. My system has evolved from a PoC to a production system and I use it daily with live data. For now, I am the only user and the data layer is pretty simple, but I plan to launch this by Q4 of 2026, which will require certain changes.

Due to its PoC nature, I dedicated most of the efforts to making sure the API was fast enough and safe enough, and the persistence layer was defined in raw SQL queries with an sqlite3 connection, something that needed improvement as well, since it was no longer good enough for a production system.

So, in order to improve this architecture, I set on a quest to migrate all my Python classes defined in one single file models.py to a production-grade design, making use of SQLAlchemy’s ORM (Object-Relational-Mapper).

ORM (Object-Relational-Mapper)

According to Wikipedia, this was first built by Oracle in September 1997. There is also an interesting talk by Anders Hejlsberg and Bruce Veckel and Bill Venners about Object-Relational Mapping in C# that goes on about the mind model of OOP and how it makes sense to treat data in the same way as objects.

In short, ORM allows data instances to be treated as objects in the programming language of your choice, letting the user skip SQL queries by performing the same operations on objects. This does not mean that SQL is no longer present, just that the way in which this is treated is different.

This increases the level of abstraction and obscures implementation details, as well as helping developers define the way in which data is accessed and controlled.

Migrating to SQLAlchemy

Creating the models

In order to migrate this data, the first step was to create a new folder called models where all these models would be stored. I created one file per model and made sure to include a __init__.py to make a module out of this directory and allow me to import it from other files without using from ./dir1/model_name.py imports.

/
└── models
    ├── __init__.py
    ├── event.py
    ├── source.py
    ├── status.py
    ├── tag.py
    └── user.py

The first step is to define a Base class. This class extends sqlalchemy.orm.DeclarativeBase and serves as the class that is extended by all the ORM’s classes that we will be defining. SQLAlchemy presents a specific, comfortable way of working, where it is possible to define where the table is coming from in the same ORM class.

class Base(DeclarativeBase):
    pass
    
# [...]

class Source(Base):
    __tablename__ = "sources" # <- table name in SQL database
    id: Mapped[int] = mapped_column(primary_key=True) # <- integer ID, using mapped_column to define the primary_key constraint.
    created_at: Mapped[datetime] = mapped_column()

After the table name, the many attributes of the class are defined. When defining a primary key, it is mandatory to let SQLAlchemy know it by using sqlalchemy.orm.mapped_column and primary_key=True. This will help the ORM retrieve data from the database.

This process was repeated with four remaining classes and the importing paths were updated to reflect the right path. Modelling the data is not a complicated task, since it only required to move from Python classes to SQLAlchemy-defined classes. The second step, replacing the SQL queries, turned out to be the interesting one.

Replacing SQL queries with CRUD

CRUD stands for the four fundamental operations that function inside DB systems: Create, Read, Update and Delete. It is also pretty much the foundation of any software system. Create info, read the info, update the info and delete the info (or mark it as deleted, and go on…).

In order for our new system to work, we need to add support for these operations, since we got rid of our raw SQL queries.

And here, a new folder comes into play, crud.

/
├── crud
└── models

Here is where we write the logic to R/W models in the database. We need to import sqlalchemy.ext.asyncio.AsyncSession to pass it on to each function as a parameter and let it use the DB session that has been passed as an argument, and also to correctly separate business logic from persistence access, making the whole system testeable.

An example goes like this, since code is worth more than a thousand words:

from models.event import Event

async def get_recents(session: AsyncSession):
    one_week_ago = datetime.now(timezone.utc) - timedelta(days=7)
    result = await session.execute(
        select(Event).where(Event.applied_at >= one_week_ago)
    events = result.scalars().all()
    return events

Disclaimer: do not take this as production code, just for educational purposes.

In the previous code snippet, it becomes clear that the session object is the way to access data. This is coupled with SQL verbs such as SELECT and WHERE. The query is performed by doing .execute and this is guarded by a very polite await (don’t forget to make your code asynchronous!). The outcome of this process returns a list of scalars that is obtained by calling result.scalars().all() and this ensures returning the whole list of items that have been applied to less than one week ago.

Here, I translated my queries into functions, such as get_recents(), create_event() and the like, and ran my pytest test suite to check that no feature regression happened.

Refactoring this code using SQLAlchemy’s ORMs was absolutely not complicated. The ease of use (after reading the docs) and the lack of complications when integrating it into a FastAPI application helps understand why it is the #1 ORM for Python. It makes it easy and uncomplicated, and allows to further modularize and separate business logic from unrelated stuff, increasing testability and application stability.