tessl/pypi-tortoise-orm

Easy async ORM for Python, built with relations in mind

Overview

Eval results

Files

Database Operations and Configuration

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

Core database management functionality through the Tortoise class and connection handling system. This includes database initialization, connection management, schema generation, and configuration options for multiple database backends.

Capabilities

Tortoise Configuration Class

Main class for configuring and managing Tortoise ORM, including database connections, model registration, and schema operations.

class Tortoise:
    """Main configuration and management class for Tortoise ORM."""
    
    apps = {}  # Registry of apps and models
    table_name_generator = None  # Custom table name generator function
    
    @classmethod
    async def init(cls,
                   config=None,
                   config_file=None,
                   _create_db=False,
                   db_url=None,
                   modules=None,
                   use_tz=False,
                   timezone="UTC",
                   routers=None,
                   table_name_generator=None):
        """
        Initialize Tortoise ORM with database connections and models.
        
        Args:
            config (dict, optional): Full configuration dictionary
            config_file (str, optional): Path to JSON/YAML config file
            db_url (str, optional): Database URL for simple setup
            modules (dict, optional): Model modules mapping for simple setup
            _create_db (bool): Create database if it doesn't exist
            use_tz (bool): Use timezone-aware datetimes
            timezone (str): Default timezone
            routers (list, optional): Database routing classes
            table_name_generator (callable, optional): Custom table name generator
            
        Raises:
            ConfigurationError: For invalid configuration
        """
    
    @classmethod
    async def generate_schemas(cls, safe=True):
        """
        Generate database schemas for all registered models.
        
        Args:
            safe (bool): Only create tables that don't exist
            
        Raises:
            ConfigurationError: If not initialized
        """
    
    @classmethod
    async def close_connections(cls):
        """
        Close all database connections cleanly.
        """
    
    @classmethod
    def describe_models(cls, models=None, serializable=True):
        """
        Describe registered models or specific models.
        
        Args:
            models (list, optional): Specific models to describe
            serializable (bool): Return JSON-serializable format
            
        Returns:
            dict: Model descriptions keyed by model name
        """
    
    @classmethod
    def init_models(cls, models_paths, app_label, _init_relations=True):
        """
        Initialize models from module paths.
        
        Args:
            models_paths (iterable): Module paths containing models
            app_label (str): Application label
            _init_relations (bool): Initialize model relationships
            
        Raises:
            ConfigurationError: For invalid models or configuration
        """
    
    @classmethod
    def get_connection(cls, connection_name):
        """
        Get database connection by name (deprecated).
        
        Args:
            connection_name (str): Connection name
            
        Returns:
            BaseDBAsyncClient: Database connection
            
        Raises:
            ConfigurationError: If connection doesn't exist
        """

Connection Management

Global connection handler for managing database connections across the application.

from tortoise import connections

# Connection handler methods
connections.get(alias)           # Get connection by name
connections.all()               # Get all connections
connections.close_all(discard=False)  # Close all connections
connections.discard(alias)      # Remove connection from registry

Database Client Interface

Base database client providing the interface for database operations.

class BaseDBAsyncClient:
    """Base async database client interface."""
    
    async def execute_query(self, query, values=None):
        """
        Execute raw SQL query.
        
        Args:
            query (str): SQL query string
            values (list, optional): Query parameters
            
        Returns:
            Result set
        """
    
    async def execute_script(self, query):
        """
        Execute SQL script.
        
        Args:
            query (str): SQL script
        """
    
    async def execute_many(self, query, values):
        """
        Execute query with multiple parameter sets.
        
        Args:
            query (str): SQL query string
            values (list): List of parameter sets
        """
    
    async def close(self):
        """Close database connection."""
    
    async def db_create(self):
        """Create database."""
    
    async def db_delete(self):
        """Delete database."""

Configuration Formats

Simple Configuration with db_url

await Tortoise.init(
    db_url='sqlite://db.sqlite3',
    modules={'models': ['app.models']}
)

Full Configuration Dictionary

config = {
    'connections': {
        'default': {
            'engine': 'tortoise.backends.asyncpg',
            'credentials': {
                'host': 'localhost',
                'port': '5432',
                'user': 'tortoise',
                'password': 'password',
                'database': 'test',
            }
        }
    },
    'apps': {
        'models': {
            'models': ['app.models'],
            'default_connection': 'default',
        }
    },
    'use_tz': False,
    'timezone': 'UTC'
}

await Tortoise.init(config=config)

JSON Configuration File

{
  "connections": {
    "default": "postgres://user:pass@localhost:5432/db"
  },
  "apps": {
    "models": {
      "models": ["app.models"],
      "default_connection": "default"
    }
  }
}

await Tortoise.init(config_file='config.json')

Database Backends

Supported database engines and their connection URLs.

SQLite

# File-based SQLite
db_url = 'sqlite://db.sqlite3'

# In-memory SQLite
db_url = 'sqlite://:memory:'

PostgreSQL

# asyncpg backend (recommended)
db_url = 'postgres://user:pass@host:port/database'

# psycopg backend
db_url = 'postgres://user:pass@host:port/database?backend=psycopg'

MySQL

db_url = 'mysql://user:pass@host:port/database'

Microsoft SQL Server

db_url = 'mssql://user:pass@host:port/database'

Oracle

db_url = 'oracle://user:pass@host:port/database'

Transaction Management

Context manager for database transactions.

from tortoise.transactions import in_transaction

async def transfer_funds(from_account, to_account, amount):
    async with in_transaction():
        from_account.balance -= amount
        to_account.balance += amount
        await from_account.save()
        await to_account.save()

# With specific connection
async with in_transaction("db_alias"):
    # Transaction operations
    pass

Usage Examples

Basic Setup

from tortoise import Tortoise
from tortoise.models import Model
from tortoise import fields

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)

async def init_db():
    # Initialize with database URL
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',
        modules={'models': ['__main__']}
    )
    
    # Generate database schema
    await Tortoise.generate_schemas()

async def close_db():
    # Clean shutdown
    await Tortoise.close_connections()

Multiple Databases

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

await Tortoise.init(config=config)

Custom Table Names

def custom_table_name(model_class):
    return f"custom_{model_class.__name__.lower()}"

await Tortoise.init(
    db_url='sqlite://db.sqlite3',
    modules={'models': ['app.models']},
    table_name_generator=custom_table_name
)

Schema Management

# Generate all schemas (fails if tables exist)
await Tortoise.generate_schemas(safe=False)

# Generate only missing schemas
await Tortoise.generate_schemas(safe=True)

# Drop all schemas (for testing)
await Tortoise._drop_databases()

Utility Functions

from tortoise import run_async

def run_async(coro):
    """
    Simple async runner that cleans up connections on exit.
    
    Args:
        coro: Coroutine to run
        
    Usage:
        async def main():
            await Tortoise.init(...)
            # Do work
            
        run_async(main())
    """

Connection Access

from tortoise import connections

# Get specific connection
conn = connections.get('default')

# Execute raw SQL
result = await conn.execute_query("SELECT COUNT(*) FROM users")

# Get all connection names
connection_names = list(connections._connections.keys())

# Close specific connection
await connections.close_all()

Install with Tessl CLI