Tortoise ORM is async-native ORM for Python. pip install tortoise-orm[asyncpg]. Define model: from tortoise.models import Model; from tortoise import fields. class User(Model): id = fields.IntField(pk=True); name = fields.CharField(max_length=100); email = fields.CharField(max_length=200, unique=True). Init: await Tortoise.init(db_url="postgres://...", modules={"models":["app.models"]}). Schema: await Tortoise.generate_schema(). Create: user = await User.create(name="Alice", email="[email protected]"). Get: user = await User.get(id=1). Get or none: user = await User.get_or_none(email="[email protected]"). Filter: users = await User.filter(is_active=True).all(). User.filter(name__contains="ali"). User.filter(created_at__gte=date). Q objects: User.filter(Q(role="admin") | Q(is_staff=True)). First: await User.filter(...).first(). Count: await User.filter(...).count(). Exists: await User.filter(...).exists(). Update: await User.filter(id=1).update(name="Bob"). Delete: await User.filter(is_active=False).delete(). Prefetch: users = await User.all().prefetch_related("orders"). Values: await User.all().values("id", "email"). Values list: await User.all().values_list("id", flat=True). Annotate: from tortoise.functions import Count; await User.all().annotate(order_count=Count("orders")).values("id","order_count"). Transaction: async with in_transaction(): .... FK: class Order(Model): user = fields.ForeignKeyField("models.User", related_name="orders"). M2M: tags = fields.ManyToManyField("models.Tag"). Migrations: pip install aerich; aerich init -t app.config.TORTOISE_ORM; aerich init-db; aerich migrate; aerich upgrade. FastAPI: register_tortoise(app, config=TORTOISE_ORM, generate_schemas=True). Testing: inherit TortoiseTestCase. Claude Code generates Tortoise ORM models, async CRUD services, and aerich migration workflows.
CLAUDE.md for Tortoise ORM
## Tortoise ORM Stack
- Version: tortoise-orm >= 0.21 | pip install "tortoise-orm[asyncpg]"
- Init: await Tortoise.init(db_url=, modules={"models":["app.models"]})
- Model: class M(Model): id=fields.IntField(pk=True)
- Query: Model.filter(**kwargs) | Model.get(pk) | Model.get_or_none()
- Create: await Model.create(**data) | obj = Model(**data); await obj.save()
- Relations: ForeignKeyField | ManyToManyField + prefetch_related
- Transactions: async with in_transaction(): ...
- Migrations: aerich init + aerich migrate + aerich upgradeTortoise ORM Async Database Pipeline
# app/models.py — Tortoise ORM model definitions
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from enum import Enum
from tortoise import fields
from tortoise.models import Model
class UserRole(str, Enum):
USER = "user"
MODERATOR = "moderator"
ADMIN = "admin"
class OrderStatus(str, Enum):
PENDING = "pending"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
# ── Models ────────────────────────────────────────────────────────────────────
class Tag(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50, unique=True)
slug = fields.CharField(max_length=50, unique=True)
class Meta:
table = "tags"
def __str__(self) -> str:
return self.name
class User(Model):
id = fields.IntField(pk=True)
email = fields.CharField(max_length=255, unique=True)
first_name = fields.CharField(max_length=100)
last_name = fields.CharField(max_length=100)
role = fields.CharEnumField(UserRole, default=UserRole.USER)
is_active = fields.BooleanField(default=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
# Reverse FK relations are available as:
# user.orders (set by Order.user FK)
class Meta:
table = "users"
indexes = [("email",), ("role", "is_active")]
def __str__(self) -> str:
return f"{self.first_name} {self.last_name} <{self.email}>"
class Product(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=200)
sku = fields.CharField(max_length=50, unique=True)
description = fields.TextField(default="")
price = fields.DecimalField(max_digits=10, decimal_places=2)
stock = fields.IntField(default=0)
is_active = fields.BooleanField(default=True)
tags = fields.ManyToManyField("models.Tag", related_name="products")
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "products"
def __str__(self) -> str:
return f"{self.sku} — {self.name}"
class Order(Model):
id = fields.IntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="orders")
status = fields.CharEnumField(OrderStatus, default=OrderStatus.PENDING)
total = fields.DecimalField(max_digits=12, decimal_places=2, default=Decimal("0"))
notes = fields.TextField(null=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "orders"
class OrderLine(Model):
id = fields.IntField(pk=True)
order = fields.ForeignKeyField("models.Order", related_name="lines")
product = fields.ForeignKeyField("models.Product", related_name="order_lines")
quantity = fields.IntField()
unit_price = fields.DecimalField(max_digits=10, decimal_places=2)
@property
def subtotal(self) -> Decimal:
return self.quantity * self.unit_price
class Meta:
table = "order_lines"# app/services.py — async CRUD and query services
from __future__ import annotations
from decimal import Decimal
from tortoise.exceptions import DoesNotExist, IntegrityError
from tortoise.functions import Avg, Count, Sum
from tortoise.queryset import Q
from tortoise.transactions import in_transaction
from app.models import Order, OrderLine, OrderStatus, Product, Tag, User, UserRole
# ── User service ──────────────────────────────────────────────────────────────
class UserService:
async def create(self, email: str, first_name: str, last_name: str,
role: UserRole = UserRole.USER) -> User:
try:
return await User.create(
email=email,
first_name=first_name,
last_name=last_name,
role=role,
)
except IntegrityError:
raise ValueError(f"User with email {email!r} already exists")
async def get_by_id(self, user_id: int) -> User:
try:
return await User.get(id=user_id)
except DoesNotExist:
raise ValueError(f"User {user_id} not found")
async def get_by_email(self, email: str) -> User | None:
return await User.get_or_none(email=email)
async def list_active(self, page: int = 1, page_size: int = 20) -> list[User]:
offset = (page - 1) * page_size
return await (
User.filter(is_active=True)
.order_by("last_name", "first_name")
.offset(offset)
.limit(page_size)
.all()
)
async def search(self, query: str) -> list[User]:
"""Search by name or email using Q objects."""
return await User.filter(
Q(first_name__icontains=query) |
Q(last_name__icontains=query) |
Q(email__icontains=query)
).filter(is_active=True).all()
async def count_by_role(self) -> list[dict]:
return await (
User.all()
.annotate(count=Count("id"))
.group_by("role")
.values("role", "count")
)
async def deactivate(self, user_id: int) -> None:
updated = await User.filter(id=user_id).update(is_active=False)
if updated == 0:
raise ValueError(f"User {user_id} not found")
# ── Order service ─────────────────────────────────────────────────────────────
class OrderService:
async def create_order(
self,
user_id: int,
lines: list[dict], # [{"product_id": int, "quantity": int}]
) -> Order:
"""
Create an order atomically:
1. Fetch user and products
2. Check stock
3. Create order + lines
4. Deduct stock
All inside a single transaction.
"""
async with in_transaction():
user = await User.get(id=user_id)
product_ids = [line["product_id"] for line in lines]
products = {p.id: p async for p in Product.filter(id__in=product_ids)}
for line_data in lines:
pid = line_data["product_id"]
if pid not in products:
raise ValueError(f"Product {pid} not found")
if products[pid].stock < line_data["quantity"]:
raise ValueError(f"Insufficient stock for product {pid}")
total = sum(
products[l["product_id"]].price * l["quantity"]
for l in lines
)
order = await Order.create(user=user, total=total)
for line_data in lines:
product = products[line_data["product_id"]]
await OrderLine.create(
order=order,
product=product,
quantity=line_data["quantity"],
unit_price=product.price,
)
await Product.filter(id=product.id).update(
stock=product.stock - line_data["quantity"]
)
return order
async def get_with_lines(self, order_id: int) -> Order:
"""Fetch order with prefetched lines and their products."""
return await (
Order.get(id=order_id)
.prefetch_related("lines__product", "user")
)
async def get_user_orders(self, user_id: int) -> list[Order]:
return await (
Order.filter(user_id=user_id)
.prefetch_related("lines")
.order_by("-created_at")
.all()
)
async def revenue_by_status(self) -> list[dict]:
return await (
Order.all()
.annotate(total_revenue=Sum("total"), order_count=Count("id"))
.group_by("status")
.values("status", "total_revenue", "order_count")
)
async def cancel(self, order_id: int) -> None:
async with in_transaction():
order = await Order.get(id=order_id).prefetch_related("lines__product")
if order.status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
raise ValueError("Cannot cancel a shipped or delivered order")
# Restore stock
for line in order.lines:
await Product.filter(id=line.product_id).update(
stock=line.product.stock + line.quantity
)
order.status = OrderStatus.CANCELLED
await order.save()# app/config.py — Tortoise init and FastAPI integration
from __future__ import annotations
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from tortoise import Tortoise
from tortoise.contrib.fastapi import register_tortoise
DATABASE_URL = os.environ.get(
"DATABASE_URL",
"sqlite://:memory:",
)
TORTOISE_ORM = {
"connections": {"default": DATABASE_URL},
"apps": {
"models": {
"models": ["app.models", "aerich.models"],
"default_connection": "default",
},
},
}
@asynccontextmanager
async def lifespan(app: FastAPI):
"""FastAPI lifespan: init Tortoise on startup, close on shutdown."""
await Tortoise.init(config=TORTOISE_ORM)
await Tortoise.generate_schema(safe=True) # safe=True: no-op if tables exist
yield
await Tortoise.close_connections()
def create_app() -> FastAPI:
app = FastAPI(lifespan=lifespan)
return app
# Alternatively use register_tortoise helper (handles lifespan automatically):
def create_app_v2() -> FastAPI:
app = FastAPI()
register_tortoise(
app,
config=TORTOISE_ORM,
generate_schemas=True,
add_exception_handlers=True,
)
return appFor the SQLAlchemy async alternative — SQLAlchemy async requires AsyncSession, async_scoped_session, separate AsyncEngine creation, and explicit await session.flush() before accessing IDs, while Tortoise ORM uses await Model.create(**data) that returns the saved instance with its id populated, async for obj in Model.filter(...): streams results without loading all in memory, and prefetch_related("lines__product") resolves two levels of FK in two queries instead of N+1. For the GINO alternative — GINO extends SQLAlchemy core with asyncio but is no longer actively maintained, while Tortoise ORM’s aerich migration tool generates Django-style numbered migration files from model diffs, class Meta: indexes and unique_together map to database-level constraints, and TortoiseTestCase wraps each test in a transaction that rolls back after the test — no test database teardown required. The Claude Skills 360 bundle includes Tortoise ORM skill sets covering Model field types, ForeignKeyField and ManyToManyField, filter/get/get_or_none/create CRUD, Q objects for complex queries, prefetch_related for eager loading, annotate with Count Sum Avg, in_transaction for atomic writes, FastAPI register_tortoise integration, aerich migration workflow, and TortoiseTestCase for isolated async tests. Start with the free tier to try async ORM code generation.