Python Tortoise ORM: Async ORM for asyncio Applications

Tortoise ORM is a Python async ORM inspired by Django ORM — it brings the familiar Meta class, ForeignKey, ManyToManyField, and queryset API to asyncio applications. It supports PostgreSQL, MySQL, MariaDB, and SQLite via async drivers, and ships with Aerich for schema migrations. Unlike SQLAlchemy's async mode (which requires more boilerplate), Tortoise ORM is async-first with a clean, concise API ideal for FastAPI projects.

Installation and Configuration

pip install tortoise-orm[asyncpg] aerich
# For MySQL: pip install tortoise-orm[asyncmy]
# For SQLite: pip install tortoise-orm  (built-in)

from tortoise import Tortoise

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.asyncpg",
            "credentials": {
                "host": "localhost",
                "port": 5432,
                "user": "myuser",
                "password": "mypassword",
                "database": "mydb",
                "minsize": 1,
                "maxsize": 20,
            },
        }
    },
    "apps": {
        "models": {
            "models": ["app.models", "aerich.models"],
            "default_connection": "default",
        }
    },
}


async def init_db():
    await Tortoise.init(config=TORTOISE_ORM)
    # Generate schemas — use only in dev; use Aerich in production
    await Tortoise.generate_schemas()


async def close_db():
    await Tortoise.close_connections()

Defining Models

from tortoise import fields
from tortoise.models import Model
from tortoise.validators import MinValueValidator, MaxValueValidator


class TimestampMixin(Model):
    """Mixin that adds created_at and updated_at to any model."""
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    class Meta:
        abstract = True


class User(TimestampMixin):
    id = fields.IntField(primary_key=True)
    email = fields.CharField(max_length=255, unique=True, index=True)
    username = fields.CharField(max_length=100, unique=True)
    password_hash = fields.CharField(max_length=255)
    role = fields.CharEnumField(enum_type=str, max_length=20, default="user")
    is_active = fields.BooleanField(default=True)
    bio = fields.TextField(null=True)

    # Reverse relations are defined automatically
    # orders: ReverseRelation[Order]

    class Meta:
        table = "users"
        indexes = [("email", "is_active"), ("role",)]

    def __str__(self):
        return f"User(id={self.id}, email={self.email})"


class Category(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100, unique=True)
    slug = fields.CharField(max_length=100, unique=True)
    parent = fields.ForeignKeyField("models.Category", null=True, related_name="children")

    class Meta:
        table = "categories"


class Product(TimestampMixin):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255, index=True)
    description = fields.TextField(default="")
    price = fields.DecimalField(max_digits=10, decimal_places=2)
    stock = fields.IntField(default=0, validators=[MinValueValidator(0)])
    is_available = fields.BooleanField(default=True)
    category = fields.ForeignKeyField("models.Category", related_name="products")
    tags = fields.ManyToManyField("models.Tag", related_name="products")

    class Meta:
        table = "products"


class Tag(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=50, unique=True)

    class Meta:
        table = "tags"

CRUD Operations

from tortoise.exceptions import DoesNotExist, IntegrityError


async def crud_examples():
    # CREATE
    user = await User.create(
        email="alice@example.com",
        username="alice",
        password_hash="hashed_password",
    )
    print(f"Created: {user}")

    # Or: get_or_create
    user, created = await User.get_or_create(
        email="alice@example.com",
        defaults={"username": "alice2", "password_hash": "hash"},
    )
    print(f"Created: {created}")

    # READ
    try:
        found = await User.get(id=1)
        found = await User.get(email="alice@example.com")
    except DoesNotExist:
        print("Not found")

    # get_or_none (returns None instead of raising)
    maybe_user = await User.get_or_none(email="nobody@example.com")

    # UPDATE — set fields and save
    found.bio = "Python developer"
    await found.save()

    # Bulk update
    await User.filter(is_active=False).update(role="archived")

    # DELETE
    await found.delete()
    await User.filter(is_active=False).delete()

Relationships

from tortoise.models import Model
from tortoise import fields


class Order(TimestampMixin):
    id = fields.IntField(primary_key=True)
    order_number = fields.CharField(max_length=50, unique=True)
    customer = fields.ForeignKeyField("models.User", related_name="orders")
    status = fields.CharField(max_length=20, default="pending")
    total = fields.DecimalField(max_digits=10, decimal_places=2, default=0)

    class Meta:
        table = "orders"


class OrderLine(Model):
    id = fields.IntField(primary_key=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)

    class Meta:
        table = "order_lines"


async def relations_examples():
    user = await User.get(id=1)

    # Create related object
    order = await Order.create(
        order_number="ORD-001",
        customer=user,
        total=99.99,
    )

    # Access ForeignKey
    await order.fetch_related("customer")
    print(order.customer.email)

    # Prefetch related (avoids N+1)
    orders = await Order.filter(
        customer__id=1
    ).prefetch_related("lines__product", "customer")

    for o in orders:
        for line in o.lines:
            print(f"{line.product.name} x{line.quantity}")

    # Many-to-many
    tag_python = await Tag.get_or_create(name="python")
    tag_async = await Tag.get_or_create(name="async")
    product = await Product.get(id=1)
    await product.tags.add(tag_python[0], tag_async[0])

    # Query through many-to-many
    python_products = await Product.filter(tags__name="python").distinct()

    # Reverse relation
    user_orders = await user.orders.filter(status="pending").all()

Advanced Querying

from tortoise.expressions import Q, F
from tortoise.functions import Count, Sum, Avg


async def advanced_queries():
    # Complex filters with Q objects (OR/AND/NOT)
    users = await User.filter(
        Q(role="admin") | Q(role="editor"),
        is_active=True,
    ).all()

    # NOT
    non_admin = await User.filter(~Q(role="admin")).all()

    # Lookups
    expensive = await Product.filter(
        price__gte=100,
        price__lte=500,
        name__icontains="keyboard",
        is_available=True,
    ).all()

    # Ordering and pagination
    recent = await Order.filter(status="pending").order_by("-created_at").offset(0).limit(20)

    # Select related with values (projection)
    email_list = await User.filter(is_active=True).values_list("email", flat=True)
    user_dicts = await User.filter(role="admin").values("id", "email", "username")

    # Annotations and aggregations
    from tortoise.functions import Count
    order_counts = await User.annotate(order_count=Count("orders")).filter(
        order_count__gte=5
    ).values("id", "email", "order_count")

    # Raw SQL when needed
    results = await User.raw("SELECT id, email FROM users WHERE created_at > $1", ["2026-01-01"])

    return users, expensive, recent, email_list

Migrations with Aerich

# pyproject.toml
# [tool.aerich]
# tortoise_orm = "app.config.TORTOISE_ORM"
# location = "./migrations"
# src_folder = "./."

# Initialize migration tracking
aerich init -t app.config.TORTOISE_ORM

# Create initial migration from current models
aerich init-db

# After changing a model, create a new migration
aerich migrate --name "add_user_bio_field"

# Apply pending migrations
aerich upgrade

# Check migration history
aerich history

# Downgrade one step
aerich downgrade

# Programmatic migration (for testing or CI)
from aerich import Command

async def run_migrations():
    command = Command(tortoise_config=TORTOISE_ORM, app="models")
    await command.init()
    await command.upgrade(run_in_transaction=True)
    print("Migrations applied")

FastAPI Integration

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tortoise import Tortoise
from tortoise.exceptions import DoesNotExist, IntegrityError
from tortoise.contrib.fastapi import RegisterTortoise


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(
        app=app,
        config=TORTOISE_ORM,
        generate_schemas=False,  # use Aerich instead
        add_exception_handlers=True,
    ):
        yield


app = FastAPI(lifespan=lifespan)


class UserCreate(BaseModel):
    email: str
    username: str
    password: str


class UserOut(BaseModel):
    id: int
    email: str
    username: str
    role: str

    class Config:
        from_attributes = True


@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(data: UserCreate):
    try:
        user = await User.create(
            email=data.email,
            username=data.username,
            password_hash=hash_password(data.password),
        )
        return UserOut.from_orm(user)
    except IntegrityError as e:
        if "email" in str(e):
            raise HTTPException(status_code=409, detail="Email already registered")
        raise HTTPException(status_code=409, detail="Username taken")


@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    try:
        user = await User.get(id=user_id)
        return UserOut.from_orm(user)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="User not found")


@app.get("/users/{user_id}/orders")
async def get_user_orders(user_id: int, status: str | None = None):
    query = Order.filter(customer_id=user_id)
    if status:
        query = query.filter(status=status)
    orders = await query.order_by("-created_at").limit(50)
    return [{"id": o.id, "order_number": o.order_number, "total": float(o.total)} for o in orders]


def hash_password(password: str) -> str:
    import hashlib
    return hashlib.sha256(password.encode()).hexdigest()

Frequently Asked Questions

Tortoise ORM vs SQLAlchemy async — which to use?

Use Tortoise ORM for projects where simplicity and a clean async-first API are priorities — especially FastAPI apps where Django-style ORM syntax is familiar. Use SQLAlchemy when you need its advanced features (session patterns, core expression language, ORM inheritance strategies) or are working with legacy SQLAlchemy code. SQLAlchemy's async support is more powerful but requires more boilerplate.

How do I run Tortoise ORM in tests?

Use tortoise.contrib.testing.TortoiseTestCase (inherits from unittest.IsolatedAsyncioTestCase) which creates an in-memory SQLite database per test. For pytest, use pytest-anyio with a fixture that calls Tortoise.init() before tests and Tortoise.close_connections() after.

Does Tortoise ORM support connection pooling?

Yes — both asyncpg (PostgreSQL) and asyncmy (MySQL) drivers support connection pooling via the minsize and maxsize credentials parameters. The pool is managed per connection name and shared across all coroutines in the same event loop.