Blending SQL and Python with SQLORM

by

At $dayjob, I use SQLAlchemy as an ORM. It’s an amazing project, powerful and flexible. However, I have always felt that some design choices are not in line with my use of ORM. Notably:

  • I’m not a big fan of the Unit of Work pattern and also the fact that you don’t have control over when the DML query is issued. I like the queries to be executed as soon as they are called in the code.
  • I don’t want my objects to be “attached” to a session or a specific database. I want to be able to fetch from one database and insert into another using the same object.
  • I want to map table rows to mostly plain objects.
  • I want to write SQL by hand for complex queries. I don’t want to use query builder or DSL, I prefer to write actual SQL.
  • I don’t care about abstracting the database. When starting a project I usually choose a database server and optimize for it.
  • Stay as close to the DB-API as possible.

With these ideas in mind, SQLORM was born. (The name is not good – I’m bad at finding names for projects like this.) It is inspired by many ORMs while bringing some unique features.

(As a side note, I know many other Python ORMs exist, but for me SQLAlchemy is the best. I don’t like others’ APIs or codebases.)

The main feature of SQLORM is that SQL is front and center. You can create SQL queries as standard Python functions using DocBlocks to write templated SQL statement:

from sqlorm import sqlfunc

@sqlfunc
def tasks_completion_report(start_date, end_date):
    """SELECT done_at, COUNT(*) count
       FROM tasks
       WHERE done_at >= %(start_date)s AND done_at <= %(end_date)s
       GROUP BY done_at"""

in this instance, start_date And end_date The parameters are there and will be saved properly. Executing the function will run the SQL query in the active transaction.

Connections and transactions are accessed through context managers. Engine Class manages DB-API connections.

from sqlorm import Engine
import datetime

engine = Engine.from_uri("sqlite://app.db")

with engine:
    report = tasks_completion_report(datetime.date(2025, 1, 1), datetime.date.today())

SQLORM provides many utilities to help you build SQL statements as well as fetch related rows in a single query.

Lines are returned as coordinates by default, but you can hydrate the object instead:

class Task:
    pass

@sqlfunc(model=Task)
def find_tasks():
    "SELECT * FROM tasks"

with engine:
    tasks = find_tasks()

Now, we don’t want to write endless simple statements to recreate basic CRUD functionality, so SQLORM provides a Model Class. It follows the active record pattern.

from sqlorm import Model

class Task(Model):
    pass

with engine:
    tasks = Task.find_all()

    task = Task.create(title="my task")

    task = Task.find_one(id=1)
    task.done = True
    task.save()

Of course, model classes can contain SQL methods!

class Task(Model):
    @classmethod
    def find_todos(cls):
        "SELECT * FROM tasks WHERE not done"

    def toggle(self):
        "UPDATE tasks SET done = not done WHERE id = %(self.id)s"

with engine:
    tasks = Task.find_todos()
    task = next(tasks)
    task.toggle()

As you’ve noticed, model classes don’t need to know the columns in advance. However, it is good practice to define columns for autocompletion, type checking, and DDL statement creation. SQLORM lets you do this using Python annotations:

from sqlorm import PrimaryKey

class Task(Model):
    id: PrimaryKey[int]
    title: str
    done: bool

Model classes provide many more utilities to handle relationships, lazy loading, column types, etc.

As mentioned earlier, model classes are not tied to any particular engine – they run on the engine provided by the current context. This makes it easier to implement patterns like reading from the replica and writing to the primary.

We can implement reading from a replica and writing to the primary in a few lines:

main = Engine.from_uri("postgresql://main")
replica = Engine.from_uri("postgresql://replica")

with replica:
    task = Task.get(1)
    if not task.done:
        with main:
            task.toggle()

SQLORM has many more powerful features. It is well documented and offers integration with Flask. Try it!



Leave a Comment