Library

Tortoise ORM

Overview

Tortoise ORM is an async ORM for Python with Django ORM-inspired API, built from scratch for asyncio. It provides optimal compatibility with async frameworks like FastAPI, Starlette, and Sanic.

Details

Tortoise ORM is designed specifically for modern Python asynchronous application development. It maintains the usability of Django ORM while achieving complete async support, demonstrating its power in real-time applications and high-throughput web service development. It supports PostgreSQL, MySQL, and SQLite, and provides excellent development experience with type hint support.

Key Features

• Fully Async: High-performance database operations through asyncio/await patterns
• Django-style API: Easy-to-learn and intuitive interface
• Type Hint Support: Modern Python development experience
• FastAPI Integration: Optimal integration with async frameworks
• Real-time Ready: Perfect compatibility with WebSockets and Server-Sent Events

Pros and Cons

Pros

• Outstanding performance in asynchronous applications
• API design familiar to Django developers
• Perfect integration with FastAPI enables modern web development
• Excellent developer experience through type hints
• Optimal for real-time application development

Cons

• Learning curve required for asynchronous programming
• Ecosystem still in development
• Limited expression for complex queries in some cases
• Difficult integration with synchronous libraries

Reference Pages

• Tortoise ORM Official Site
• Tortoise ORM Documentation
• Tortoise ORM GitHub

Code Examples

Installation and Basic Setup

# Install Tortoise ORM
pip install tortoise-orm

# Database drivers
pip install asyncpg  # PostgreSQL
pip install aiomysql  # MySQL
pip install aiosqlite  # SQLite

# For FastAPI integration
pip install fastapi tortoise-orm[fastapi]

# settings.py
TORTOISE_ORM = {
    "connections": {
        "default": "postgres://user:password@localhost:5432/mydb"
        # or SQLite: "sqlite://db.sqlite3"
        # or MySQL: "mysql://user:password@localhost:3306/mydb"
    },
    "apps": {
        "models": {
            "models": ["app.models", "aerich.models"],
            "default_connection": "default",
        },
    },
    "use_tz": False,
    "timezone": "UTC"
}

# Application initialization
from tortoise import Tortoise

async def init_db():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

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

Basic CRUD Operations (Model Definition, Create, Read, Update, Delete)

# models.py
from tortoise.models import Model
from tortoise import fields
from typing import Optional

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    
    # Relations
    posts: fields.ReverseRelation["Post"]
    
    class Meta:
        table = "users"
        
    def __str__(self):
        return self.name

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    published_at = fields.DatetimeField(null=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    
    # Foreign key
    author: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
        "models.User", related_name="posts"
    )
    
    class Meta:
        table = "posts"
        
    def __str__(self):
        return self.title

# Run migrations
# aerich init -t settings.TORTOISE_ORM
# aerich init-db
# aerich migrate

# Basic CRUD operations
async def crud_examples():
    # Create
    user = await User.create(
        name="John Doe",
        email="[email protected]"
    )
    
    # Read
    all_users = await User.all()
    user_by_id = await User.get(id=1)
    user_by_email = await User.get(email="[email protected]")
    
    # Multiple condition search
    active_users = await User.filter(is_active=True)
    
    # Update
    user = await User.get(id=1)
    user.name = "John Smith"
    await user.save()
    
    # Bulk update
    await User.filter(is_active=False).update(is_active=True)
    
    # Delete
    user = await User.get(id=1)
    await user.delete()
    
    # Bulk delete
    await User.filter(is_active=False).delete()
    
    return user

Advanced Queries and Relationships

from tortoise.query_utils import Q
from tortoise.functions import Count, Max, Min, Avg
from datetime import datetime, timedelta

async def advanced_queries():
    # Complex conditional queries
    recent_active_users = await User.filter(
        Q(is_active=True) & 
        Q(created_at__gte=datetime.now() - timedelta(days=30)) &
        Q(email__icontains="@company.com")
    ).order_by("-created_at").limit(10)
    
    # Relationship queries (Prefetch)
    users_with_posts = await User.all().prefetch_related("posts")
    
    for user in users_with_posts:
        print(f"{user.name}: {len(user.posts)} posts")
    
    # Conditional relation queries
    users_with_recent_posts = await User.filter(
        posts__created_at__gte=datetime.now() - timedelta(days=7)
    ).distinct().prefetch_related("posts")
    
    # Aggregation queries
    user_stats = await User.annotate(
        post_count=Count("posts")
    ).filter(post_count__gt=0)
    
    # Subqueries and JOINs
    popular_authors = await User.filter(
        posts__published_at__isnull=False
    ).annotate(
        published_posts=Count("posts", _filter=Q(posts__published_at__isnull=False)),
        latest_post=Max("posts__created_at")
    ).filter(published_posts__gte=5)
    
    # Complex search
    search_results = await Post.filter(
        Q(title__icontains="Python") | Q(content__icontains="Python")
    ).select_related("author")
    
    # Aggregate functions
    stats = await Post.all().aggregate(
        total_posts=Count("id"),
        avg_content_length=Avg("content__length"),
        latest_post=Max("created_at")
    )
    
    return recent_active_users, stats

# Custom QuerySet
class UserQuerySet:
    def __init__(self, model_class):
        self.model_class = model_class
    
    def active(self):
        return self.model_class.filter(is_active=True)
    
    def with_posts(self):
        return self.model_class.filter(posts__isnull=False).distinct()
    
    def recent(self, days=30):
        cutoff = datetime.now() - timedelta(days=days)
        return self.model_class.filter(created_at__gte=cutoff)

# Usage example
async def custom_queries():
    queryset = UserQuerySet(User)
    active_users_with_posts = await queryset.active().with_posts().recent(7)

Migrations and Schema Management

# Using Aerich migration tool
pip install aerich

# Initialize
aerich init -t settings.TORTOISE_ORM

# Create initial database
aerich init-db

# Generate migrations
aerich migrate --name "add_user_status"

# Apply migrations
aerich upgrade

# Downgrade
aerich downgrade

# Check current migration status
aerich heads

# Programmatic schema generation
from tortoise import Tortoise

async def create_tables():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

# Custom migration
async def custom_migration():
    from tortoise import connections
    
    conn = connections.get("default")
    
    # Execute custom SQL
    await conn.execute_query("""
        CREATE INDEX IF NOT EXISTS idx_users_email_active 
        ON users (email, is_active)
    """)
    
    await conn.execute_query("""
        CREATE INDEX IF NOT EXISTS idx_posts_author_published 
        ON posts (author_id, published_at) 
        WHERE published_at IS NOT NULL
    """)

# Data seeding
async def seed_data():
    # Create test data
    users_data = [
        {"name": "User 1", "email": "[email protected]"},
        {"name": "User 2", "email": "[email protected]"},
        {"name": "User 3", "email": "[email protected]"},
    ]
    
    users = await User.bulk_create([User(**data) for data in users_data])
    
    # Create related data
    posts_data = []
    for user in users:
        for i in range(3):
            posts_data.append(Post(
                title=f"{user.name}'s Post {i+1}",
                content=f"This is a post by {user.name}.",
                author=user,
                published_at=datetime.now() if i % 2 == 0 else None
            ))
    
    await Post.bulk_create(posts_data)

Performance Optimization and Advanced Features

from tortoise.transactions import in_transaction, atomic
from tortoise.expressions import F

async def performance_optimization():
    # Using transactions
    async with in_transaction() as connection:
        user = await User.create(
            name="Jane Doe",
            email="[email protected]",
            using_db=connection
        )
        
        post = await Post.create(
            title="First Post",
            content="My first post using Tortoise ORM",
            author=user,
            published_at=datetime.now(),
            using_db=connection
        )
    
    # Atomic operations (using decorator)
    @atomic()
    async def create_user_with_posts(name: str, email: str, post_titles: list):
        user = await User.create(name=name, email=email)
        
        posts = []
        for title in post_titles:
            posts.append(Post(
                title=title,
                content=f"Content for {title}",
                author=user
            ))
        
        await Post.bulk_create(posts)
        return user
    
    # Batch operations
    await User.bulk_create([
        User(name="User 1", email="[email protected]"),
        User(name="User 2", email="[email protected]"),
        User(name="User 3", email="[email protected]"),
    ])
    
    # Batch update
    await User.filter(is_active=False).update(is_active=True)
    
    # Atomic updates with F expressions
    await Post.filter(id=1).update(view_count=F("view_count") + 1)
    
    # Connection pool configuration
    TORTOISE_ORM = {
        "connections": {
            "default": {
                "engine": "tortoise.backends.asyncpg",
                "credentials": {
                    "host": "localhost",
                    "port": 5432,
                    "user": "postgres",
                    "password": "password",
                    "database": "mydb",
                    "minsize": 1,
                    "maxsize": 10,
                    "max_queries": 50000,
                    "max_inactive_connection_lifetime": 300,
                }
            }
        },
        "apps": {
            "models": {
                "models": ["app.models"],
                "default_connection": "default",
            }
        }
    }

# Custom fields
class TimestampMixin:
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

class User(Model, TimestampMixin):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    
    # Custom property
    @property
    def display_name(self) -> str:
        return f"{self.name} ({self.email})"
    
    # Custom method
    async def get_recent_posts(self, days: int = 30) -> list:
        cutoff = datetime.now() - timedelta(days=days)
        return await self.posts.filter(created_at__gte=cutoff)

Framework Integration and Practical Examples

# FastAPI integration
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel
from typing import List

app = FastAPI()

# Pydantic models
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    created_at: datetime
    
    class Config:
        orm_mode = True

class PostResponse(BaseModel):
    id: int
    title: str
    content: str
    published_at: Optional[datetime]
    author: UserResponse
    
    class Config:
        orm_mode = True

# API endpoints
@app.get("/users/", response_model=List[UserResponse])
async def get_users(skip: int = 0, limit: int = 10):
    users = await User.all().offset(skip).limit(limit)
    return users

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

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    try:
        db_user = await User.create(**user.dict())
        return db_user
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/users/{user_id}/posts/", response_model=List[PostResponse])
async def get_user_posts(user_id: int):
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    posts = await Post.filter(author=user).select_related("author")
    return posts

# WebSocket integration
from fastapi import WebSocket, WebSocketDisconnect
import json

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # Real-time database operations
            message_data = json.loads(data)
            
            if message_data["type"] == "create_post":
                post = await Post.create(
                    title=message_data["title"],
                    content=message_data["content"],
                    author_id=message_data["author_id"]
                )
                
                # Broadcast new post to all clients
                await manager.broadcast(json.dumps({
                    "type": "new_post",
                    "post": {
                        "id": post.id,
                        "title": post.title,
                        "content": post.content,
                        "created_at": post.created_at.isoformat()
                    }
                }))
                
    except WebSocketDisconnect:
        manager.disconnect(websocket)

# Background tasks
from fastapi import BackgroundTasks

async def send_notification_email(user_id: int):
    user = await User.get(id=user_id)
    # Email sending logic
    print(f"Sending notification to {user.email}")

@app.post("/users/{user_id}/notify/")
async def notify_user(user_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_notification_email, user_id)
    return {"message": "Notification scheduled"}

# Tortoise ORM configuration
register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
    add_exception_handlers=True,
)

# Starlette integration
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def list_users(request):
    users = await User.all()
    return JSONResponse([{
        "id": user.id,
        "name": user.name,
        "email": user.email
    } for user in users])

async def create_user(request):
    data = await request.json()
    user = await User.create(**data)
    return JSONResponse({
        "id": user.id,
        "name": user.name,
        "email": user.email
    })

starlette_app = Starlette(routes=[
    Route("/users", list_users, methods=["GET"]),
    Route("/users", create_user, methods=["POST"]),
])

# Application startup initialization
@app.on_event("startup")
async def startup_event():
    await init_db()
    await seed_data()  # Only in development environment

@app.on_event("shutdown")
async def shutdown_event():
    await close_db()

# Test integration
import pytest
from tortoise.contrib.test import finalizer, initializer

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="module", autouse=True)
async def initialize_tests(request):
    await initializer(["app.models"])
    request.addfinalizer(finalizer)

@pytest.mark.asyncio
async def test_create_user():
    user = await User.create(name="Test User", email="[email protected]")
    assert user.name == "Test User"
    assert user.email == "[email protected]"
    assert user.is_active is True

@pytest.mark.asyncio
async def test_user_posts_relationship():
    user = await User.create(name="Author", email="[email protected]")
    post = await Post.create(
        title="Test Post",
        content="Test content",
        author=user
    )
    
    await user.fetch_related("posts")
    assert len(user.posts) == 1
    assert user.posts[0].title == "Test Post"