Pyndatic Query - SQL query builder with pydantic & pypika combined

https://github.com/pydantic/pydantic https://github.com/kayak/pypika

Usage

Define your tables

from pydantic_query import DbModel, relationship

# User is a Pydantic model and a pypika Table
class User(DbModel):
    __tablename__ = 'users'

    id: int
    login: str
    email: str | None = None
    created_dt: datetime | None = None

    balance: "Balance" = relationship("id", "user_id")


class Balance(DbModel):
    __tablename__ = 'balances'

    user_id: int
    amount: Decimal

    user: "User" = relationship("user_id", "id")

Create database session

from pydantic_query.postgres import AsyncSession
session = await AsyncSession.connect(settings.DATABASE_URL)

Use it to query

from pydantic_query import select

# fetch list of Pydantic objects, single object, or any kind of dict or list

all_users: list[User] = await session.fetch_all(select(User))
user: User | None = await session.fetch_one(select(User.id, User.login).where(User.login == 'nik'))
users_by_id: dict[int, User] = await session.fetch_dict(select(User.id, User))
user_ids: list[int] = await session.fetch_all(select(User.id))
user_to_login: dict[int, str] = await session.fetch_kv(select(User.id, User.login)) # fetch key/value

Insert

from pydantic_query import insert

await session.insert(User(login='new_user'))

# bulk insert
await session.insert([User(login=f'new_user_{i}') for i in range(10)])

# upsert
await session.insert(
    Balance(user_id=user.id, amount=amount))
    .on_conflict(Balance.user_id)
    .do_update(amount=Balance.amount + amount)
)

Update, Delete

new_balance = await session.update(Balance).where(Balance.user_id == user.id).set(Balance.amount, Balance.amount + add_amount).returning(Balance)
old_balance = await session.delete(Balance).where(Balance.user_id == user.id).returning(Balance)

Join

No lazy fetching. Accessing non-joined field is an error. relationship() is simply marks the field as a relationship.

user  = await session.fetch_one(select(User))
print(user.balance.amount) # raises RelationshipNotJoined

user = await session.fetch_one(select(User).join(Balance).on(User.id == Balance.user_id)).where(User.id == 1)
print(user.balance.amount) # works

In the example above we have only one field of type Balance on user, so join knows which field to fill. What if you have multiple foreign keys of same type?

class Image(DbModel):
    id: int
    path: str
    creator_id: int
    editor_id: int | None

    created_by: "User" = relationship("creator_id", "id")
    edited_by: "User | None" = relationship("editor_id", "id")

image = await session.fetch_one(select(Image).join(User).on(Image.creator_id == User.id).where(User.id == user_id))
print(image.created_by.login) # works
print(image.edited_by.login) # raises, not fetched

Aggregation

from pydantic_query import select
from pypika import functions as fn

# select u.id, count(i.id) as user_images from users u join images i on i.creator_id == u.id group by u.id
user_image_count = await session.fetch_kv(select(User.id, fn.Count(Image.id)).group_by(User.id))

CTE

Common Table Expressions (CTEs) work via the .with_() method:

from pydantic_query import select
from pypika.terms import ValueWrapper

# CTE with pydantic_query models
balance_cte = select(Balance.user_id, Balance.amount).where(Balance.amount > 50)
query = select(Balance.user_id, Balance.amount).with_(balance_cte, "high_balances").from_(balance_cte)

# ValueWrapper for literal values in select
t_user = Table("users")
query = session.QueryCls.from_(t_user).select(t_user.id, ValueWrapper(42).as_("constant_value"))

Transactions

By default, each query executes and commits automatically. For more control, you can use explicit transactions.

Auto-commit (default)

Every execute() call automatically commits after execution:

await session.execute("INSERT INTO users (name) VALUES ('nik')")  # auto-commits

Transaction context manager

Use session.transaction() to wrap multiple operations in a transaction. It commits on success and rolls back on exception:

async with await session.transaction():
    await session.execute("INSERT INTO accounts (id, balance) VALUES (1, 100)")
    await session.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
# automatically commits; rolls back if exception occurs

Savepoints

Use savepoint=True for named savepoints (generates random name) or pass a string for custom name:

async with await session.transaction(savepoint=True):
    await session.execute("...")
    await session.rollback()  # rolls back to savepoint
    # continues after rollback

async with await session.transaction(savepoint="my_sp"):
    await session.execute("...")

Manual begin/commit/rollback

For full control, use session.begin():

tx = await session.begin()
try:
    await session.execute("INSERT INTO orders (user_id) VALUES (1)")
    await tx.commit()
except:
    await tx.rollback()
    raise

One-to-many and many-to-many

class User(DbModel):
    __tablename__ = 'users'

    id: int

    # one-to-many
    images: list["Image"] = relationship("id", "user_id")


class Image(DbModel):
    __tablename__ = 'images'

    id: int
    user_id: int

    # many-to-many
    tags: list["Tag"] = many_to_many("ImageTags")


class Tag(DbModel):
    __tablename__ = 'tags'

    id: int
    tag: str


class ImageTags(DbModel):
    __tablename__ = 'tags_images'
    
    image_id: int
    tag_id: int


users = await session.fetch_one(
    select(User)
    .join(Image).on(User.id == Image.user_id)
    .join(ImageTags).on(ImageTags.image_id == Image.id)
    .join(Tag).on(Tag.id == ImageTags.tag_id)
)
print(users.images[0].tags[0].tag)

Non-goals

What this library is not aimed for:

• ORM. This is a query builder. There will be no session.refresh, session.add and session.expunge. Changing fields of the objects will not change values in database
• Migrations. Schema management is outside of the scope of this project. You are advised to use any existing migration tools such as goose https://github.com/pressly/goose