tessl/pypi-tortoise-orm

Easy async ORM for Python, built with relations in mind

Overview

Eval results

Files

Framework Integration

Name: tessl/pypi-tortoise-orm
Author: tessl

Integration utilities and extensions for web frameworks, serialization libraries, and testing utilities. Tortoise ORM provides first-class support for popular Python web frameworks and tools, making it easy to integrate into existing applications.

Capabilities

FastAPI Integration

Seamless integration with FastAPI through middleware and utilities.

from tortoise.contrib.fastapi import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False,
    add_exception_handlers=False
):
    """
    Register Tortoise ORM with FastAPI application.
    
    Args:
        app: FastAPI application instance
        config (dict, optional): Tortoise configuration
        config_file (str, optional): Path to config file
        db_url (str, optional): Database URL
        modules (dict, optional): Model modules
        generate_schemas (bool): Auto-generate database schemas
        add_exception_handlers (bool): Add ORM exception handlers
    """

def HTTPNotFoundError():
    """HTTP 404 exception for FastAPI."""

def register_tortoise_middlewares(app):
    """Register Tortoise middlewares with FastAPI app."""

Pydantic Integration

Generate Pydantic models from Tortoise models for API serialization.

from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator

def pydantic_model_creator(
    cls,
    name=None,
    exclude=(),
    include=(),
    computed=(),
    allow_cycles=True,
    sort_alphabetically=False,
    exclude_readonly=False,
    meta_override=None,
    validators=None,
    module=None
):
    """
    Create Pydantic model from Tortoise model.
    
    Args:
        cls: Tortoise model class
        name (str, optional): Name for Pydantic model
        exclude (tuple): Fields to exclude
        include (tuple): Fields to include (if specified, only these)
        computed (tuple): Computed fields to include
        allow_cycles (bool): Allow circular references
        sort_alphabetically (bool): Sort fields alphabetically
        exclude_readonly (bool): Exclude readonly fields
        meta_override (dict, optional): Override Meta attributes
        validators (dict, optional): Field validators
        module (str, optional): Module name for the model
        
    Returns:
        Type[BaseModel]: Pydantic model class
    """

def pydantic_queryset_creator(
    cls,
    name=None,
    exclude=(),
    include=(),
    computed=(),
    allow_cycles=True,
    sort_alphabetically=False
):
    """
    Create Pydantic model for queryset serialization.
    
    Args:
        cls: Tortoise model class
        name (str, optional): Name for Pydantic model
        exclude (tuple): Fields to exclude
        include (tuple): Fields to include
        computed (tuple): Computed fields to include
        allow_cycles (bool): Allow circular references
        sort_alphabetically (bool): Sort fields alphabetically
        
    Returns:
        Type[BaseModel]: Pydantic model for lists
    """

Testing Utilities

Testing helpers and database setup utilities for unit and integration tests.

from tortoise.contrib.test import TestCase, TortoiseTestCase, initializer, finalizer

class TestCase:
    """Base test case with Tortoise ORM setup."""
    pass

class TortoiseTestCase(TestCase):
    """Async test case with automatic Tortoise setup/teardown."""
    
    async def asyncSetUp(self):
        """Setup method called before each test."""
    
    async def asyncTearDown(self):
        """Teardown method called after each test."""

def initializer(modules, db_url=None, loop=None):
    """
    Initialize Tortoise for testing.
    
    Args:
        modules (list): Model modules to load
        db_url (str, optional): Database URL (defaults to in-memory SQLite)
        loop: Event loop to use
    """

def finalizer():
    """Clean up Tortoise after testing."""

def getDBConfig(app_label="models", modules=None):
    """
    Get database configuration for testing.
    
    Args:
        app_label (str): Application label
        modules (list, optional): Model modules
        
    Returns:
        dict: Database configuration
    """

Framework Middleware

Middleware integrations for various web frameworks.

Starlette/FastAPI Middleware

from tortoise.contrib.starlette import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False
):
    """Register Tortoise with Starlette/FastAPI app."""

Sanic Integration

from tortoise.contrib.sanic import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False
):
    """Register Tortoise with Sanic app."""

Quart Integration

from tortoise.contrib.quart import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False
):
    """Register Tortoise with Quart app."""

aiohttp Integration

from tortoise.contrib.aiohttp import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False
):
    """Register Tortoise with aiohttp app."""

BlackSheep Integration

from tortoise.contrib.blacksheep import register_tortoise

def register_tortoise(
    app,
    config=None,
    config_file=None,
    db_url=None,
    modules=None,
    generate_schemas=False
):
    """Register Tortoise with BlackSheep app."""

Database-Specific Extensions

Extensions providing database-specific functionality.

PostgreSQL Extensions

from tortoise.contrib.postgres.fields import TSVectorField, ArrayField
from tortoise.contrib.postgres.functions import Random, ArrayAgg, JsonAgg

class TSVectorField:
    """PostgreSQL text search vector field."""
    def __init__(self, **kwargs): ...

class ArrayField:
    """PostgreSQL array field."""
    def __init__(self, element_type, size=None, **kwargs): ...

# PostgreSQL-specific functions
def Random(): ...
def ArrayAgg(field, distinct=False, ordering=None): ...
def JsonAgg(field, distinct=False, ordering=None): ...

MySQL Extensions

from tortoise.contrib.mysql.fields import GeometryField
from tortoise.contrib.mysql.functions import GroupConcat, Rand

class GeometryField:
    """MySQL geometry field."""
    def __init__(self, srid=4326, **kwargs): ...

# MySQL-specific functions
def GroupConcat(field, separator=',', distinct=False, order_by=None): ...
def Rand(): ...

Usage Examples

FastAPI Integration Example

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from tortoise.models import Model
from tortoise import fields
from tortoise.contrib.pydantic import pydantic_model_creator

# Define model
class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    email = fields.CharField(max_length=100, unique=True)
    
    class Meta:
        table = "users"

# Create Pydantic models
User_Pydantic = pydantic_model_creator(User, name="User")
UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", exclude_readonly=True)

# FastAPI app
app = FastAPI()

# Register Tortoise
register_tortoise(
    app,
    db_url="sqlite://db.sqlite3",
    modules={"models": ["__main__"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

@app.post("/users/", response_model=User_Pydantic)
async def create_user(user: UserIn_Pydantic):
    user_obj = await User.create(**user.dict(exclude_unset=True))
    return await User_Pydantic.from_tortoise_orm(user_obj)

@app.get("/users/{user_id}", response_model=User_Pydantic)
async def get_user(user_id: int):
    return await User_Pydantic.from_tortoise_orm(
        await User.get(id=user_id)
    )

Pydantic Model Generation

from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.models import Model
from tortoise import fields

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    content = fields.TextField()
    author = fields.ForeignKeyField("models.User", related_name="posts")
    created_at = fields.DatetimeField(auto_now_add=True)

# Create different Pydantic models for different use cases
PostIn = pydantic_model_creator(
    Post, 
    name="PostIn",
    exclude_readonly=True,  # Exclude id, created_at
    exclude=("author",)     # Exclude author for input
)

PostOut = pydantic_model_creator(
    Post,
    name="PostOut", 
    include=("id", "title", "created_at", "author.name")  # Include author name
)

PostList = pydantic_queryset_creator(Post, name="PostList")

# Usage in API
@app.post("/posts/", response_model=PostOut)
async def create_post(post_data: PostIn, current_user: User):
    post = await Post.create(**post_data.dict(), author=current_user)
    return await PostOut.from_tortoise_orm(post)

Testing Setup

from tortoise.contrib.test import TestCase
from tortoise.contrib.test import initializer, finalizer
import asyncio

class UserTestCase(TestCase):
    async def asyncSetUp(self):
        # Set up test data
        self.user = await User.create(
            name="Test User",
            email="test@example.com"
        )
    
    async def test_user_creation(self):
        user = await User.create(
            name="Alice",
            email="alice@example.com"
        )
        self.assertEqual(user.name, "Alice")
        self.assertEqual(user.email, "alice@example.com")
    
    async def test_user_query(self):
        users = await User.filter(name__contains="Test").all()
        self.assertEqual(len(users), 1)
        self.assertEqual(users[0].name, "Test User")

# Manual test setup
async def test_setup():
    await initializer(["app.models"])
    
    # Run tests
    try:
        # Test code here
        user = await User.create(name="Test", email="test@example.com")
        assert user.name == "Test"
    finally:
        await finalizer()

# Run test
asyncio.run(test_setup())

Multiple Framework Integration

# Sanic example
from sanic import Sanic
from tortoise.contrib.sanic import register_tortoise

app = Sanic("MyApp")

register_tortoise(
    app,
    db_url="postgres://user:pass@localhost/db",
    modules={"models": ["app.models"]},
    generate_schemas=True
)

# Quart example
from quart import Quart
from tortoise.contrib.quart import register_tortoise

app = Quart(__name__)

register_tortoise(
    app,
    config={
        'connections': {
            'default': 'sqlite://db.sqlite3'
        },
        'apps': {
            'models': {
                'models': ['app.models'],
                'default_connection': 'default',
            }
        }
    }
)

Database-Specific Features

# PostgreSQL array field
from tortoise.contrib.postgres.fields import ArrayField
from tortoise.contrib.postgres.functions import ArrayAgg

class Article(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    tags = ArrayField(fields.CharField(max_length=50))

# Query with array functions
articles_with_tags = await Article.annotate(
    all_tags=ArrayAgg('tags')
).all()

# MySQL geometry field
from tortoise.contrib.mysql.fields import GeometryField

class Location(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    coordinates = GeometryField()

Custom Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from tortoise import connections

class DatabaseMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Custom database logic before request
        response = await call_next(request)
        # Custom database logic after request
        return response

# Add to FastAPI
app.add_middleware(DatabaseMiddleware)

Error Handling Integration

from fastapi import HTTPException
from tortoise.exceptions import DoesNotExist, IntegrityError

@app.exception_handler(DoesNotExist)
async def does_not_exist_handler(request, exc):
    raise HTTPException(status_code=404, detail="Resource not found")

@app.exception_handler(IntegrityError)
async def integrity_error_handler(request, exc):
    raise HTTPException(status_code=400, detail="Data integrity violation")

Install with Tessl CLI