ORM

Ushka ships an async ORM built on SQLAlchemy 2.0 with a small Active Record interface: models know how to save, delete and query themselves. Migrations are handled by Alembic through ushka db.

Installing

The ORM is an optional extra, so the core install stays small:

pip install "ushka[orm]"        # SQLAlchemy, Alembic and the SQLite driver (aiosqlite)
pip install "ushka[postgres]"   # + asyncpg
pip install "ushka[mysql]"      # + aiomysql

Choosing the database

The URL comes from the configuration, in this order of precedence: environment / .env (USHKA_DATABASE_URL, or DATABASE_URL, or the older USHKA_DB_URL), then [database] url in ushka.toml, then the default sqlite:///ushka.db.

# ushka.toml
[database]
url = "sqlite:///shop.db"

# .env, keep credentials out of ushka.toml
USHKA_DATABASE_URL=postgresql://user:password@localhost/shop

Write plain URLs: sqlite://, postgresql:// (or postgres://) and mysql:// are switched to their async drivers (aiosqlite, asyncpg, aiomysql) automatically. A URL that already names a driver (postgresql+psycopg://...) is used as is.

A relative SQLite path such as sqlite:///shop.db is relative to the directory the process runs in, so start the server (and ushka db) from the project directory, or use an absolute path: sqlite:////srv/shop/shop.db.

The engine is created lazily on first use. To point the ORM somewhere else from code (in tests, for example) call db.configure(url) before the first query.

Defining models

# models.py
import enum

from ushka.orm import db


class Status(enum.Enum):
    DRAFT = "draft"
    PUBLISHED = "published"


class User(db.Model):
    __tablename__ = "users"

    name = db.String(length=100)
    email = db.String(unique=True)
    is_active = db.Bool(default=True)
    created_at = db.DateTime(auto_now=True)
    posts = db.Relationship("Post", back_populates="author")


class Post(db.Model):
    __tablename__ = "posts"

    title = db.String(length=200)
    body = db.Text(nullable=True)
    status = db.Enum(Status, default=Status.DRAFT)
    tags = db.JSON(default=list)
    author_id = db.ForeignKey("users.id")
    author = db.Relationship("User", back_populates="posts")

Every model needs a __tablename__. Do not declare id: db.Model already provides an auto-incrementing BigInteger primary key named id.

Column helpers

Columns are NOT NULL unless you pass nullable=True.

db.Model is a regular SQLAlchemy declarative base, so plain SQLAlchemy Column(...) definitions work as well.

Creating the tables: migrations

Tables are created and changed through migrations. Alembic has to see your models, and it looks for them by importing app.models, app or main from the project directory, so import your models in app.py:

# app.py
from ushka import Ushka

import models  # noqa: F401  (registers the models for `ushka db make`)

app = Ushka()

Then, from the project directory:

ushka db init                       # once: creates migrations/ and alembic.ini
ushka db make -m "create users"     # generates a migration from your models
ushka db migrate                    # applies it

Every time you change a model, run ushka db make -m "..." and ushka db migrate again. ushka db revert undoes the last migration. Review the generated file in migrations/versions/ before applying it. The full list of commands is in the CLI guide.

SQLite migrations run in Alembic's batch mode automatically, so columns can be altered and dropped there too.

CRUD

All database methods are coroutines: remember the await.

Create

user = await User.create(name="Alice", email="alice@example.com")

bob = User(name="Bob", email="bob@example.com")
await bob.save()

Read

users = await User.all()                     # list
user = await User.get(1)                     # by primary key, or None
alice = await User.first(name="Alice")       # first match, or None
active = await User.filter(is_active=True)   # list of matches

first() and filter() take keyword arguments and compare for equality. For anything richer, build a SQLAlchemy statement with Model.select() and run it with db.execute():

result = await db.execute(
    User.select().where(User.email.endswith("@example.com")).order_by(User.name)
)
users = result.scalars().all()

Update

user = await User.get(1)
if user:
    user.name = "Alicia"
    await user.save()

Delete

user = await User.first(name="Bob")
if user:
    await user.delete()

Serialising

instance.to_dict() returns the columns as a dict (enums become their value, UUIDs become strings). Dates and times stay datetime objects, so convert them before returning the dict as JSON:

data = user.to_dict()
data["created_at"] = user.created_at.isoformat()
return data

Request data and the ORM

Values from the request are Untrusted strings. The ORM methods and SQLAlchemy expressions always send them as bound parameters, so they need no special care, but convert types yourself:

# routes/users/[user_id].py
from ushka.guard import safe
from ushka.http.exceptions import HttpNotFound

from models import User


async def get(user_id):
    user = await User.get(safe.integer(user_id, min=1))
    if user is None:
        raise HttpNotFound(message="No such user")
    return {"id": user.id, "name": user.name}

Raw SQL

db.execute() accepts a SQLAlchemy construct, a SQL string, or (on Python 3.14+) a t-string. SQL strings are guarded: if any part came from the request, the call is refused with USHKA-G005. Pass values as parameters instead:

# :name placeholders, values bound by the driver
result = await db.execute(
    db.text("SELECT id, name FROM users WHERE email = :email", email=email)
)
rows = result.all()

# Python 3.14+: every {value} in a t-string becomes a bound parameter
result = await db.execute(t"SELECT id, name FROM users WHERE email = {email}")

Use db.text() rather than sqlalchemy.text(): SQLAlchemy copies the string internally and the guard would lose track of where it came from.

Relationships and sessions

The Active Record methods open a short session per call, so the objects they return are detached. Accessing a relationship that was not loaded (post.author) then raises DetachedInstanceError. Load it up front:

from sqlalchemy.orm import selectinload

result = await db.execute(Post.select().options(selectinload(Post.author)))
for post in result.scalars().all():
    print(post.title, post.author.name)

For several operations in one transaction, open a session yourself:

async with db.Session() as session:
    session.add_all([User(name="Ann", email="ann@example.com"),
                     User(name="Ben", email="ben@example.com")])
    await session.commit()

db.engine (the AsyncEngine) and db.metadata are available for anything else SQLAlchemy can do.