Examples

All examples are located in the examples/ directory and can be run directly:

PYTHONPATH=. .venv/bin/python examples/basic_usage.py

Basic Usage

import asyncio

from metaorm import BaseFilter, BaseRepository, BaseTable, Field, RepositorySettings

class UserTable(BaseTable, table=True): tablename = "users"

id: int | None = Field(default=None, primary_key=True)
name: str
email: str = Field(unique=True)

class UserFilter(BaseFilter): name: str | None = None email: str | None = None

class UserRepository(BaseRepository, table=UserTable, filter_=UserFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") repository = UserRepository(settings=settings)

await repository.create_tables()

# Create
user = await repository.create_item(
    UserTable(name="Alice", email="alice@example.com"),
)
print(f"Created: {user.name}, {user.email}")

# Read single item
single = await repository.get_item()
print(f"Single item: {single.name}, {single.email}")

# Read by filter
filtered = await repository.get_item(
    filter_=UserFilter(email="alice@example.com"),
)
print(f"Filtered item: {filtered.name if filtered else None}")

# Read all
items = [item async for item in repository.get_items()]
print(f"All items: {[(item.name, item.email) for item in items]}")

# Count
count = await repository.get_items_count()
print(f"Count: {count}")

# Update
updated = [item async for item in repository.update_items(name="Updated")]
print(f"Updated: {[(item.name, item.email) for item in updated]}")

# Delete
await repository.delete_items()
count = await repository.get_items_count()
print(f"Count after delete: {count}")

if name == "main": asyncio.run(main())

DTO Usage

import asyncio

from pydantic import BaseModel

from metaorm import BaseFilter, BaseRepository, BaseTable, Field, RepositorySettings

class User(BaseModel): id: int | None = None name: str email: str

class UserTable(BaseTable[User], table=True): tablename = "users"

id: int | None = Field(default=None, primary_key=True)
name: str
email: str = Field(unique=True)

@classmethod
def from_item(cls, item: User) -> "UserTable":
    return cls(id=item.id, name=item.name, email=item.email)

def to_item(self) -> User:
    return User(id=self.id, name=self.name, email=self.email)

class UserFilter(BaseFilter): name: str | None = None email: str | None = None

class UserRepository(BaseRepository, table=UserTable, filter_=UserFilter, dto=User): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") repository = UserRepository(settings=settings)

await repository.create_tables()

# Create using DTO
user = await repository.create_item(
    User(name="Alice", email="alice@example.com"),
)
print(f"Created DTO: {user.model_dump()}")

# Read single DTO
single = await repository.get_item(
    filter_=UserFilter(email="alice@example.com"),
)
print(f"Single DTO: {single.model_dump() if single else None}")

# Read all — returned as DTOs
items = [item async for item in repository.get_items()]
print(f"Items as DTOs: {[item.model_dump() for item in items]}")

if name == "main": asyncio.run(main())

Transactions

import asyncio

from metaorm import BaseFilter, BaseRepository, BaseTable, Field, RepositorySettings

class ProductTable(BaseTable, table=True): tablename = "products"

id: int | None = Field(default=None, primary_key=True)
name: str
price: float

class ProductFilter(BaseFilter): name: str | None = None price: int | None = None

class ProductRepository(BaseRepository, table=ProductTable, filter_=ProductFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") repository = ProductRepository(settings=settings)

await repository.create_tables()

# Explicit transaction via repository
async with repository.transaction():
    product1 = await repository.create_item(
        ProductTable(name="Laptop", price=999.99),
    )
    product2 = await repository.create_item(
        ProductTable(name="Mouse", price=29.99),
    )
    print(f"Created in transaction: {product1.name}, {product2.name}")

# Reusing an existing session (no new savepoint)
async with repository.transaction(), repository.transaction():
    items = [item async for item in repository.get_items()]
    print(f"Items in reused session: {len(items)}")

# True nested transaction (savepoint) via repository
try:
    async with repository.nested_transaction():
        await repository.create_item(
            ProductTable(name="Keyboard", price=79.99),
        )
        raise ValueError("Rollback nested")
except ValueError:
    pass

count = await repository.get_items_count()
print(f"Items after nested rollback: {count}")  # 2

# Read single item inside a transaction
async with repository.transaction():
    item = await repository.get_item(filter_=ProductFilter(name="Laptop"))
    print(f"Single in transaction: {item.name if item else None}")

if name == "main": asyncio.run(main())

Nested Transactions

import asyncio

from metaorm import BaseFilter, BaseRepository, BaseTable, Field, RepositorySettings

class ProductTable(BaseTable, table=True): tablename = "products"

id: int | None = Field(default=None, primary_key=True)
name: str
price: float

class ProductFilter(BaseFilter): name: str | None = None

class ProductRepository(BaseRepository, table=ProductTable, filter_=ProductFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") repository = ProductRepository(settings=settings)

await repository.create_tables()

# Standalone nested transaction: rollback only the inner scope
try:
    async with repository.nested_transaction():
        await repository.create_item(ProductTable(name="Laptop", price=999.99))
        await repository.create_item(ProductTable(name="Mouse", price=29.99))
        raise ValueError("Simulated error inside nested transaction")
except ValueError:
    pass

count = await repository.get_items_count()
print(f"Items after standalone nested rollback: {count}")  # 0

# Nested transaction inside an outer transaction
async with repository.transaction():
    await repository.create_item(ProductTable(name="Keyboard", price=79.99))

    try:
        async with repository.nested_transaction():
            await repository.create_item(
                ProductTable(name="Monitor", price=299.99),
            )
            raise ValueError("Nested rollback")
    except ValueError:
        pass

    # Monitor is rolled back, Keyboard stays in the outer transaction
    items = [item async for item in repository.get_items()]
    print(f"Items after partial rollback: {len(items)}")  # 1
    print(items[0].name)  # Keyboard

# Verify committed results
all_items = [item async for item in repository.get_items()]
print(f"Final items: {[item.name for item in all_items]}")  # ["Keyboard"]

if name == "main": asyncio.run(main())

Filters, Pagination & Sorting

import asyncio

from metaorm import ( BaseFilter, BaseRepository, BaseSort, BaseTable, Field, OffsetPagination, RepositorySettings, )

class BookTable(BaseTable, table=True): tablename = "books"

id: int | None = Field(default=None, primary_key=True)
title: str
year: int

class BookFilter(BaseFilter): title: str | None = None year: int | None = None

class BookRepository(BaseRepository, table=BookTable, filter_=BookFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") repository = BookRepository(settings=settings)

await repository.create_tables()

# Seed data
for index in range(10):
    await repository.create_item(
        BookTable(title=f"Book {index}", year=2020 + index),
    )

# Get single item by filter
year_filter = BookFilter(year=2025)
single = await repository.get_item(filter_=year_filter)
print(f"Single year = 2025: {single.title if single else None}")

# Filter by year = 2025
filtered = [item async for item in repository.get_items(filter_=year_filter)]
print(f"Year = 2025: {[book.title for book in filtered]}")

# Filter by exact title
title_filter = BookFilter(title="Book 5")
filtered = [item async for item in repository.get_items(filter_=title_filter)]
print(f"Title = 'Book 5': {[book.title for book in filtered]}")

# Pagination
pagination = OffsetPagination(offset=2, limit=3)
page = [item async for item in repository.get_items(pagination=pagination)]
print(f"Page (offset=2, limit=3): {[book.title for book in page]}")

# Sorting descending
sort = BaseSort(sort_by="year", sort_by_order="desc")
sorted_items = [item async for item in repository.get_items(sort=sort)]
print(f"Sorted by year desc: {[book.year for book in sorted_items]}")

if name == "main": asyncio.run(main())

Eager Loading

import asyncio

from sqlalchemy.orm import joinedload

from metaorm import ( BaseFilter, BaseRepository, BaseTable, Field, Relationship, RepositoriesContainer, RepositorySettings, )

class AuthorTable(BaseTable, table=True): tablename = "authors"

id: int | None = Field(default=None, primary_key=True)
name: str
books: list["BookTable"] = Relationship(back_populates="author")

class BookTable(BaseTable, table=True): tablename = "books"

id: int | None = Field(default=None, primary_key=True)
title: str
author_id: int = Field(foreign_key="authors.id")
author: AuthorTable = Relationship(back_populates="books")

class BookFilter(BaseFilter): title: str | None = None author_id: int | None = None

class AuthorFilter(BaseFilter): name: str | None = None

class BookRepository(BaseRepository, table=BookTable, filter_=BookFilter): pass

class AuthorRepository(BaseRepository, table=AuthorTable, filter_=AuthorFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") container = RepositoriesContainer(settings=settings)

author_repo = AuthorRepository(container=container)
book_repo = BookRepository(container=container)

await author_repo.create_tables()
await book_repo.create_tables()

# Create author and books
author = await author_repo.create_item(AuthorTable(name="Tolkien"))
await book_repo.create_item(BookTable(title="The Hobbit", author_id=author.id))
await book_repo.create_item(BookTable(title="LOTR", author_id=author.id))

# Load books with authors using joinedload
books = [
    item
    async for item in book_repo.get_items(
        options=[joinedload(BookTable.author)],
    )
]
for book in books:
    print(f"Book: {book.title}, Author: {book.author.name}")

if name == "main": asyncio.run(main())

Multi-Repo Transactions

import asyncio

from metaorm import ( BaseFilter, BaseRepository, BaseTable, Field, RepositoriesContainer, RepositorySettings, )

class UserTable(BaseTable, table=True): tablename = "users"

id: int | None = Field(default=None, primary_key=True)
name: str

class OrderTable(BaseTable, table=True): tablename = "orders"

id: int | None = Field(default=None, primary_key=True)
user_id: int
total: float

class UserFilter(BaseFilter): name: str | None = None

class OrderFilter(BaseFilter): user_id: int | None = None

class UserRepository(BaseRepository, table=UserTable, filter_=UserFilter): pass

class OrderRepository(BaseRepository, table=OrderTable, filter_=OrderFilter): pass

async def main() -> None: settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:") container = RepositoriesContainer(settings=settings)

user_repo = container.get_repository(UserRepository)
order_repo = container.get_repository(OrderRepository)

# Create tables for all repositories at once via the container
await container.create_tables(UserRepository, OrderRepository)

# Or create tables individually per repository
# await user_repo.create_tables()
# await order_repo.create_tables()

# Atomic transaction across two repositories
async with container.transaction():
    user = await user_repo.create_item(UserTable(name="Alice"))
    await order_repo.create_item(OrderTable(user_id=user.id, total=100.00))
    await order_repo.create_item(OrderTable(user_id=user.id, total=250.50))

# Nested transaction inside outer transaction (savepoint)
async with container.transaction():
    user = await user_repo.create_item(UserTable(name="Bob"))
    try:
        async with container.nested_transaction():
            await order_repo.create_item(
                OrderTable(user_id=user.id, total=999.99),
            )
            raise ValueError("Rollback nested order")
    except ValueError:
        pass
    # Bob stays, the order is rolled back

# Verify results
users = [item async for item in user_repo.get_items()]
orders = [item async for item in order_repo.get_items()]

print(f"Users: {[user.name for user in users]}")
print(f"Orders: {[(order.user_id, order.total) for order in orders]}")
print(f"Orders count: {len(orders)}")

# Get single user by name
single_user = await user_repo.get_item(filter_=UserFilter(name="Alice"))
print(f"Single user: {single_user.name if single_user else None}")

if name == "main": asyncio.run(main())