Alchemical, the unknown friend of SQLAlchemy

Simplify your SQLAlchemy workflow with Alchemical

Photo by Growtika on Unsplash

Did you struggle to correctly instantiate a database session with SQLAlchemy? Managing migrations with alembic? Handling multiple databases? I had at least struggled with the first two and I’m betting I’m not alone.

This is where Alchemical comes in. It will radically simplify your life for the above-mentioned problems.

Installation

You will need python 3.7+ to install it. Also, note that Alchemical works with SQLAlchemy 1.4+.

$ pip install alchemical
# or if you use poetry
$ poetry add alchemical

If you don’t know poetry, I have a tutorial on it.

What is new in poetry 1.2

I highly recommend it to handle your project dependencies.

Usage

Create a database connection

You just need to pass the connection string to the constructor of the Alchemical class.

db = Alchemical('sqlite:///file.db')

If you need to pass options as you will normally do using the create_engine function, you can use the Alchemical engine_options argument and pass a dictionary.

db = Alchemical('sqlite:///file.db', engine_options={'echo': True})

Define database models

The Alchemical instance contains the base model you will need to define all your database models. Here I use the new SQLAlchemy syntax introduced in version 2.0. More information can be found in their quickstart tutorial.

from datetime import datetime
from typing import Optional

from alchemical import Alchemical
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Boolean, DateTime, func
db = Alchemical('sqlite:///file.db')

class Todo(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    description: Mapped[Optional[str]]
    done: Mapped[bool] = mapped_column(Boolean(), default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=func.now(), onupdate=func.now())

Create and destroy the database

It is as simple as this:

# Creates tables that don't already exist in the database
db.create_all()
# Drops the database
db.drop_all()

Note that you probably don’t want to use these commands in a production environment. To create and manage database migrations, you will probably want to use a tool like Alembic. We will talk about it later in this article.

Get a database connection and make queries

Here is how you can get a scoped database session using the with statement. I don’t put the whole code with the model declaration, just the relevant part.

from sqlalchemy import select

with db.Session() as session:
    todo = Todo(name='foo', description='bar')
    session.add(todo)
    session.commit()
    query = select(Todo).where(Todo.done.is_(False))
    for todo in session.scalars(query):
        print(todo)

Notes:

• The session object we get with the context manager is the one we are used to when using SQLAlchemy directly.
• Here again, I use the new syntax introduction in SQLAlchemy 1.4+. I will probably write another article on how to use it in a modern web application project so stay tuned. 😉
• In SQLAlchemy 2.0+ the session is not bound to a thread anymore and can have a shorter lifecycle. This is a more correct approach in my opinion.

There is a variant to the previous construction which is ideal if you deal with transactions. In the following code, and particularly the first with statement, if one operation (insertion in our case) fails, the whole block is rollback. And there is no need to commit at the end, it is automatically done. To check we use another session to query all the objects created.

with db.begin() as session:
    t1 = Todo(name='1', description='1')
    t2 = Todo(name='2', description='2')
    session.add_all([t1, t2])

with db.Session() as session:
    for todo in session.scalars(select(Todo)):
        print(todo.name)

One flaw with Alchemical is that the code is not typed, so we lost auto-completion for the session object although typing was added in the new version of SQLAlchemy. 😕

Asyncio support

Another goodie brought by SQLAlchemy is the asyncio support! Now we can easily integrate it in asynchronous environments such as async web frameworks. Yes, I’m thinking of frameworks like FastAPI. With this, we can write more robust applications handling a lot of IO operations. Alchemical just leverages it without additional complexity.

To run the following code, you will need to install the aiosqlite package.

import asyncio
from alchemical.aio import Alchemical

db = Alchemical('sqlite:///file.db')
# model creation does not change
async def main():
    await db.drop_all()
    await db.create_all()
    async with db.begin() as session:
        t1 = Todo(name='1', description='1')
        t2 = Todo(name='2', description='2')
        session.add_all([t1, t2])
    async with db.Session() as session:
        for todo in await session.scalars(select(Todo)):
            print(todo.name)
asyncio.run(main())

Support multiple databases

Alchemical makes it simple to support multiple databases using one database instance. There are two ways to accomplish this.

The first way is to define a bind key representing a database connection on each model we create.

db = Alchemical(binds={
    'users': 'postgresql://user:password@localhost/mydb',
    'cache': 'sqlite:///',
})

class User(db.Model):
    __bind_key__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    nickname: Mapped[str] = mapped_column(String(50))

Here the User model is bound to the PostgreSQL database.

The second way is to define a default database and bind keys for the other databases. This way, all models that don’t have a bind key will be automatically bound to the default database.

db = Alchemical('mysqldb://user:password@localhost/db', binds={
    'users': 'postgresql://user:password@localhost/mydb',
    'cache': 'sqlite:///',
})

class Todo(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    description: Mapped[Optional[str]]
    done: Mapped[bool] = mapped_column(Boolean(), default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=func.now(), onupdate=func.now())

class User(db.Model):
    __bind_key__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    nickname: Mapped[str] = mapped_column(String(50))

Here the Todo model is bound to the MySQL database and the User model is bound to the PostgreSQL database.

Database migrations with Alembic

One of my favorite features is the integration with Alembic. I have always struggled to configure it properly with SQLAlchemy, but Alchemical makes it a breeze.

You first need to install Alembic and then initialize its environment with a special Alchemical command.

$ pip install alembic
$ python -m alchemical.alembic.cli init migrations

Normally you should have a migrations folder and an alembic.ini file created. If you open the alembic.ini file, you should see a line like the following:

alchemical_db = main:db

On this line, the location of the Alchemical instance is set. So if you have a package my_package and inside it, a module named models.py where you instantiate your database instance called db, you will need to define it like this:

alchemical_db = my_package.models:db

After that you can use Alembic, as usual, to create migrations and update the database.

$ alembic revision --autogenerate -m "Initial migration"
$ alembic upgrade head

Easy peasy!

This is all for this article, hope you enjoy reading it. If you want to know more about Alchemical, check the official documentation.

Take care of yourself and see you soon! 😁

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter on substack 😉