embar 0.2.0

A SQL-y and well-typed ORM for Python

Embar

A Python ORM with types

---

Embar is a new ORM for Python with the following goals:

• Type safety: your type checker should know what arguments are valid, and what is being returned from any call.
• Type hints: your LSP should be able to guide you towards the query you want to write.
• SQL-esque: you should be able to write queries simply by knowing SQL and your data model.
• You should be able to actually just write SQL when you need to.

These are mostly inspired by Drizzle. The Python ecosystem deserves something with similar DX.

Embar supports three database clients:

• SQLite 3 via the Python standard library
• Postgres via psycopg3
• Postgres via async psycopg3

The async psycopg3 client is recommended. The others are provided mostly for testing and experimenting locally.

Embar uses Template strings and so only supports Python 3.14.

Embar is pre-alpha and ready for experimentation but not production use.

Documentation: embar.rdrn.me

Quickstart

The quickstart uses the non-async sqlite client to make an easy example.

Install

uv add embar

Set up database models

# schema.py
from embar.column.common import Integer, Text
from embar.config import EmbarConfig
from embar.table import Table

class User(Table):
    # If you don't provide a table name, it is generated from your class name
    embar_config = EmbarConfig(table_name="users")

    id: Integer = Integer(primary=True)
    # Columns will also generate their own name if not provided
    email: Text = Text("user_email", default="text", not_null=True)

class Message(Table):
    id: Integer = Integer()
    # Foreign key constraints are easy to add
    user_id: Integer = Integer().fk(lambda: User.id)
    content: Text = Text()

Create client and apply migrations

In production, you would (probably) use the embar CLI to generate and run migrations. This example uses the utility function to do it all in code.

# main.py
import sqlite3
from embar.db.sqlite import Db as SqliteDb

conn = sqlite3.connect(":memory:")
db = SqliteDb(conn)
db.migrate([User, Message]).run()

Insert some data

user = User(id=1, email="[e-mail]")
message = Message(id=1, user_id=user.id, content="Hello!")

db.insert(User).values(user).run()
db.insert(Message).values(message).run()

Query some data

With join, where and group by.

from typing import Annotated
from embar.query.selection import Selection
from embar.query.where import Eq, Like, Or

class UserSel(Selection):
    id: Annotated[int, User.id]
    messages: Annotated[list[str], Message.content.many()]

users = (
    db.select(UserSel)
    .fromm(User)
    .left_join(Message, Eq(User.id, Message.user_id))
    .where(Or(
        Eq(User.id, 1),
        Like(User.email, "foo%")
    ))
    .group_by(User.id)
    .run()
)
# [ UserSel(id=1, messages=['Hello!']) ]

Query some more data

This time with fully nested child tables, and some raw SQL.

from datetime import datetime
from embar.sql import Sql

class UserHydrated(Selection):
    email: Annotated[str, User.email]
    messages: Annotated[list[Message], Message.many()]
    date: Annotated[datetime, Sql(t"CURRENT_TIMESTAMP")]

users = (
    db.select(UserHydrated)
    .fromm(User)
    .left_join(Message, Eq(User.id, Message.user_id))
    .group_by(User.id)
    .limit(2)
    .run()
)
# [UserHydrated(
#      email='[e-mail]',
#      messages=[Message(content='Hello!', id=1, user_id=1)],
#      date: datetime(2025, 10, 26, ...)
# )]

See the SQL

Every query produces exactly one... query. And you can always see what's happening under the hood with the .sql() method:

users_query = (
    db.select(UserHydrated)
    .fromm(User)
    .left_join(Message, Eq(User.id, Message.user_id))
    .group_by(User.id)
    .sql()
)
users_query.sql
# SELECT 
#     "users"."user_email" AS "email",
#     json_group_array(json_object(
#         'id', "message"."id",
#         'user_id', "message"."user_id",
#         'content', "message"."content"
#     )) AS "messages",
#     CURRENT_TIMESTAMP AS "date"
# FROM "users"
# LEFT JOIN "message" ON "users"."id" = "message"."user_id"
# GROUP BY "users"."id"

Update a row

Unfortunately this requires another model to be defined, as Python doesn't have a Partial[] type.

from typing import TypedDict

class MessageUpdate(TypedDict, total=False):
    id: int
    user_id: int
    content: str

(
    db.update(Message)
    .set(MessageUpdate(content="Goodbye"))
    .where(Eq(Message.id, 1))
    .run()
)

Add indexes

from embar.constraint import Index

class Message(Table):
    embar_config: EmbarConfig = EmbarConfig(
        constraints=[Index("message_idx").on(lambda: Message.user_id)]
    )
    user_id: Integer = Integer().fk(lambda: User.id)

Run raw SQL

db.sql(t"DELETE FROM {Message}").run()

Or with a return:

class UserId(Selection):
    id: Annotated[int, int]

res = (
    db.sql(t"SELECT * FROM {User}")
    .model(UserId)
    .run()
)
# [UserId(id=1)]

Migrations

Properly diffing migrations is not supported yet, but it's in the pipeline.

In the meantime, you have two options:

Embar CLI (work in progress)

This uses which uses an LLM (and your ANTHROPIC_API_KEY) to generate vibe-diffs. You should inspect these before running them.

You can see a working example at example/.

First create a config file embar.yml in your app root:

dialect: postgresql
db_url: postgresql://user:password@localhost:5432/db
schema_path: app.schema
migrations_dir: migrations   # optional

Then to generate migrations, run the following and follow the prompts:

embar migrate

Or to push directly to your db, run the following. You will be prompted before each change.

embar push

Or use an external schema management tool

Use the migrate() method shown in the quickstart to dump the current DDL to a .sql file.

Then use a schema management tool to manage updates. Some options are:

• Atlas
• sqldef
• sqitch

Contributing

Install uv.

Then:

uv sync

This project uses poethepoet for tasks/scripts.

You'll need Docker installed to run tests.

Format, lint, type-check, test:

uv run poe fmt
           lint
           check
           test

# or
uv run poe all

Or do this:

# Run this or put it in .zshrc/.bashrc/etc
alias poe="uv run poe"

# Then you can just:
poe test

Other ORMs to consider

There seems to be a gap in the Python ORM market.

• SQLAlchemy (and, by extension, SQLModel) is probably great if you're familiar with it, but too complicated for people who don't live in it
• PonyORM has no types
• PugSQL has no types
• TortoiseORM is probably appealing if you like Django/ActiveRecord
• Piccolo is cool but not very type-safe (functions accept Any, return dicts)
• ormar is not very type-safe and still based on SQLAlchemy