tessl/pypi-tortoise-orm

Easy async ORM for Python, built with relations in mind

Overview

Eval results

Files

Error Handling

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

Comprehensive exception hierarchy for handling configuration errors, database operational errors, integrity constraints, validation failures, and query-related exceptions. Tortoise ORM provides specific exceptions for different error scenarios to enable proper error handling and debugging.

Capabilities

Exception Hierarchy

Base exception classes and their inheritance structure.

class BaseORMException(Exception):
    """
    Base ORM exception class.
    
    All Tortoise ORM exceptions inherit from this base class.
    """

class ConfigurationError(BaseORMException):
    """
    Configuration error exception.
    
    Raised when there are issues with Tortoise ORM configuration,
    such as invalid database settings, missing models, or incorrect
    field definitions.
    """

class FieldError(BaseORMException):
    """
    Field definition error exception.
    
    Raised when there is a problem with model field definitions,
    such as invalid field types, incorrect field options, or
    conflicting field configurations.
    """

class ParamsError(BaseORMException):
    """
    Parameter error exception.
    
    Raised when functions cannot be executed with the provided
    parameters, such as invalid query parameters or incorrect
    function arguments.
    """

class TransactionManagementError(BaseORMException):
    """
    Transaction management error exception.
    
    Raised when there are issues with database transaction
    management, such as nested transaction conflicts or
    transaction state errors.
    """

Database Operation Exceptions

Exceptions related to database operations and constraints.

class OperationalError(BaseORMException):
    """
    Database operational error exception.
    
    Raised when database operations fail due to operational
    issues such as connection problems, query execution errors,
    or database-specific errors.
    """

class IntegrityError(OperationalError):
    """
    Database integrity constraint violation exception.
    
    Raised when database operations violate integrity constraints
    such as unique constraints, foreign key constraints, or
    check constraints.
    """

class ValidationError(BaseORMException):
    """
    Model validation error exception.
    
    Raised when model field validation fails, such as invalid
    field values, constraint violations, or custom validation
    errors.
    """

Query-Related Exceptions

Exceptions specific to query operations and results.

class DoesNotExist(BaseORMException):
    """
    Object does not exist exception.
    
    Raised when a query expects to find a single object but
    no matching object exists in the database.
    """

class MultipleObjectsReturned(BaseORMException):
    """
    Multiple objects returned exception.
    
    Raised when a query expects to find a single object but
    multiple matching objects exist in the database.
    """

class ObjectDoesNotExistError(BaseORMException):
    """
    Object does not exist error exception.
    
    Alternative exception for cases where objects are expected
    to exist but are not found.
    """

class IncompleteInstanceError(BaseORMException):
    """
    Incomplete model instance error exception.
    
    Raised when trying to perform operations on model instances
    that are missing required data or are in an incomplete state.
    """

Usage Examples

Configuration Error Handling

from tortoise import Tortoise
from tortoise.exceptions import ConfigurationError

async def setup_database():
    try:
        await Tortoise.init(
            config={
                'connections': {
                    'default': 'invalid://connection/string'
                },
                'apps': {
                    'models': {
                        'models': ['nonexistent.models'],
                    }
                }
            }
        )
    except ConfigurationError as e:
        print(f"Configuration error: {e}")
        # Handle configuration errors
        # - Check database connection string
        # - Verify model module paths
        # - Validate configuration structure

Database Operation Error Handling

from tortoise.exceptions import IntegrityError, OperationalError

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

async def create_user(email, name):
    try:
        user = await User.create(email=email, name=name)
        return user
    except IntegrityError as e:
        print(f"Integrity constraint violated: {e}")
        # Handle constraint violations
        # - Unique constraint violations
        # - Foreign key constraint violations
        # - Check constraint violations
        raise ValueError("Email already exists")
    except OperationalError as e:
        print(f"Database operation failed: {e}")
        # Handle operational errors
        # - Connection timeouts
        # - Database locks
        # - Query execution errors
        raise RuntimeError("Database temporarily unavailable")

Query Error Handling

from tortoise.exceptions import DoesNotExist, MultipleObjectsReturned

async def get_user_by_email(email):
    try:
        user = await User.get(email=email)
        return user
    except DoesNotExist:
        print(f"No user found with email: {email}")
        return None
    except MultipleObjectsReturned:
        print(f"Multiple users found with email: {email}")
        # This shouldn't happen with unique email field
        # but could occur with non-unique fields
        users = await User.filter(email=email).all()
        return users[0]  # Return first one or handle differently

async def get_user_by_id(user_id):
    try:
        user = await User.get(id=user_id)
        return user
    except DoesNotExist:
        raise ValueError(f"User with ID {user_id} not found")

Validation Error Handling

from tortoise.exceptions import ValidationError

class Product(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    price = fields.DecimalField(max_digits=10, decimal_places=2)
    
    def clean(self):
        if self.price < 0:
            raise ValidationError("Price cannot be negative")

async def create_product(name, price):
    try:
        product = Product(name=name, price=price)
        await product.full_clean()  # Run validation
        await product.save()
        return product
    except ValidationError as e:
        print(f"Validation failed: {e}")
        # Handle validation errors
        # - Invalid field values
        # - Custom validation failures
        # - Field constraint violations
        raise ValueError(f"Invalid product data: {e}")

Transaction Error Handling

from tortoise.exceptions import TransactionManagementError
from tortoise.transactions import in_transaction

async def transfer_money(from_account, to_account, amount):
    try:
        async with in_transaction():
            # Debit from source account
            from_account.balance -= amount
            await from_account.save()
            
            # Credit to target account  
            to_account.balance += amount
            await to_account.save()
            
    except TransactionManagementError as e:
        print(f"Transaction failed: {e}")
        # Handle transaction errors
        # - Deadlocks
        # - Transaction conflicts
        # - Rollback failures
        raise RuntimeError("Money transfer failed")
    except Exception as e:
        print(f"Unexpected error during transaction: {e}")
        # Transaction will be automatically rolled back
        raise

Comprehensive Error Handling

from tortoise.exceptions import (
    BaseORMException, ConfigurationError, FieldError, 
    OperationalError, IntegrityError, ValidationError,
    DoesNotExist, MultipleObjectsReturned
)

class UserService:
    @staticmethod
    async def create_user(email, name, age=None):
        try:
            # Validate input
            if not email or '@' not in email:
                raise ValidationError("Invalid email address")
            
            if not name or len(name.strip()) == 0:
                raise ValidationError("Name is required")
            
            # Create user
            user = await User.create(
                email=email.lower().strip(),
                name=name.strip(),
                age=age
            )
            return user
            
        except ValidationError as e:
            # Re-raise validation errors with context
            raise ValueError(f"User validation failed: {e}")
            
        except IntegrityError as e:
            # Handle database constraint violations
            if "unique" in str(e).lower():
                raise ValueError("Email address already exists")
            else:
                raise ValueError("User data violates database constraints")
                
        except OperationalError as e:
            # Handle database operational errors
            raise RuntimeError(f"Database error during user creation: {e}")
            
        except BaseORMException as e:
            # Catch any other ORM-specific errors
            raise RuntimeError(f"ORM error: {e}")
    
    @staticmethod
    async def get_user(user_id=None, email=None):
        try:
            if user_id:
                return await User.get(id=user_id)
            elif email:
                return await User.get(email=email)
            else:
                raise ParamsError("Either user_id or email must be provided")
                
        except DoesNotExist:
            return None
            
        except MultipleObjectsReturned:
            # This shouldn't happen with unique fields
            raise RuntimeError("Data integrity issue: multiple users found")
            
        except ParamsError as e:
            raise ValueError(str(e))
    
    @staticmethod
    async def update_user(user_id, **updates):
        try:
            user = await User.get(id=user_id)
            
            # Apply updates
            for field, value in updates.items():
                if hasattr(user, field):
                    setattr(user, field, value)
                else:
                    raise FieldError(f"Unknown field: {field}")
            
            await user.save()
            return user
            
        except DoesNotExist:
            raise ValueError(f"User with ID {user_id} not found")
            
        except FieldError as e:
            raise ValueError(str(e))
            
        except ValidationError as e:
            raise ValueError(f"Update validation failed: {e}")
            
        except IntegrityError as e:
            raise ValueError(f"Update violates constraints: {e}")

Error Logging and Monitoring

import logging
from tortoise.exceptions import BaseORMException

logger = logging.getLogger(__name__)

async def safe_database_operation(operation_func, *args, **kwargs):
    """Wrapper for safe database operations with logging."""
    try:
        return await operation_func(*args, **kwargs)
        
    except ConfigurationError as e:
        logger.error(f"Database configuration error: {e}")
        raise
        
    except IntegrityError as e:
        logger.warning(f"Data integrity violation: {e}")
        raise
        
    except OperationalError as e:
        logger.error(f"Database operational error: {e}")
        raise
        
    except DoesNotExist as e:
        logger.info(f"Object not found: {e}")
        raise
        
    except ValidationError as e:
        logger.warning(f"Validation error: {e}")
        raise
        
    except BaseORMException as e:
        logger.error(f"Unexpected ORM error: {e}")
        raise
        
    except Exception as e:
        logger.error(f"Unexpected error in database operation: {e}")
        raise

# Usage
result = await safe_database_operation(User.create, email="test@example.com", name="Test User")

Web Framework Error Handling

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

async def get_user_endpoint(user_id: int):
    try:
        user = await User.get(id=user_id)
        return user
    except DoesNotExist:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

async def create_user_endpoint(user_data: dict):
    try:
        user = await User.create(**user_data)
        return user
    except ValidationError as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Validation error: {e}"
        )
    except IntegrityError as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="User with this email already exists"
        )

Install with Tessl CLI