tessl/pypi-tortoise-orm

Easy async ORM for Python, built with relations in mind

Overview

Eval results

Files

Querying and Filtering

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

Advanced query building capabilities with QuerySet, Q expressions, filtering, ordering, aggregation, and bulk operations for efficient database access. Tortoise ORM provides a rich query API similar to Django's ORM but optimized for async operations.

Capabilities

QuerySet Operations

Lazy query builder that supports method chaining for complex database queries.

class QuerySet:
    """Lazy query builder for database queries."""
    
    def filter(self, **kwargs):
        """
        Filter queryset by given criteria.
        
        Args:
            **kwargs: Field lookups (field__lookup=value)
            
        Returns:
            QuerySet: Filtered queryset
        """
    
    def exclude(self, **kwargs):
        """
        Exclude records matching criteria.
        
        Args:
            **kwargs: Field lookups to exclude
            
        Returns:
            QuerySet: Filtered queryset
        """
    
    def order_by(self, *fields):
        """
        Order queryset by given fields.
        
        Args:
            *fields: Field names, prefix with '-' for descending
            
        Returns:
            QuerySet: Ordered queryset
        """
    
    def limit(self, limit):
        """
        Limit number of results.
        
        Args:
            limit (int): Maximum number of results
            
        Returns:
            QuerySet: Limited queryset
        """
    
    def offset(self, offset):
        """
        Skip number of results.
        
        Args:
            offset (int): Number of results to skip
            
        Returns:
            QuerySet: Offset queryset
        """
    
    def distinct(self):
        """
        Return distinct results.
        
        Returns:
            QuerySet: Distinct queryset
        """
    
    def select_related(self, *fields):
        """
        Follow foreign keys and select related data.
        
        Args:
            *fields: Related field names to select
            
        Returns:
            QuerySet: Queryset with related data
        """
    
    def prefetch_related(self, *fields):
        """
        Prefetch related objects in separate queries.
        
        Args:
            *fields: Related field names to prefetch
            
        Returns:
            QuerySet: Queryset with prefetched data
        """
    
    def annotate(self, **kwargs):
        """
        Add annotations to queryset.
        
        Args:
            **kwargs: Annotations (name=expression)
            
        Returns:
            QuerySet: Annotated queryset
        """
    
    def group_by(self, *fields):
        """
        Group results by fields.
        
        Args:
            *fields: Field names to group by
            
        Returns:
            QuerySet: Grouped queryset
        """
    
    async def all(self):
        """
        Get all results as a list.
        
        Returns:
            list: All matching model instances
        """
    
    async def first(self):
        """
        Get first result or None.
        
        Returns:
            Model or None: First matching instance
        """
    
    async def get(self, **kwargs):
        """
        Get single result matching criteria.
        
        Args:
            **kwargs: Filter criteria
            
        Returns:
            Model: Matching model instance
            
        Raises:
            DoesNotExist: If no match found
            MultipleObjectsReturned: If multiple matches found
        """
    
    async def get_or_none(self, **kwargs):
        """
        Get single result or None.
        
        Args:
            **kwargs: Filter criteria
            
        Returns:
            Model or None: Matching instance or None
        """
    
    async def count(self):
        """
        Count matching results.
        
        Returns:
            int: Number of matching records
        """
    
    async def exists(self):
        """
        Check if any results exist.
        
        Returns:
            bool: True if results exist
        """
    
    async def delete(self):
        """
        Delete all matching records.
        
        Returns:
            int: Number of deleted records
        """
    
    async def update(self, **kwargs):
        """
        Update all matching records.
        
        Args:
            **kwargs: Fields to update
            
        Returns:
            int: Number of updated records
        """

Q Expressions

Complex query expressions for advanced filtering with AND, OR, and NOT operations.

class Q:
    """Query expression for complex conditions."""
    
    def __init__(self, **kwargs):
        """
        Create Q expression.
        
        Args:
            **kwargs: Field lookups
        """
    
    def __and__(self, other):
        """AND operation with another Q expression."""
    
    def __or__(self, other):
        """OR operation with another Q expression."""
    
    def __invert__(self):
        """NOT operation (negation)."""

# Usage with Q expressions
from tortoise.queryset import Q

# Complex conditions
queryset = Model.filter(
    Q(name__startswith='A') | Q(age__gte=18)
)

queryset = Model.filter(
    Q(status='active') & ~Q(name__in=['admin', 'root'])
)

Field Lookups

Available field lookup types for filtering and querying.

# Exact match
Model.filter(name='Alice')
Model.filter(name__exact='Alice')

# Case-insensitive exact match
Model.filter(name__iexact='alice')

# Contains
Model.filter(name__contains='Ali')
Model.filter(name__icontains='ali')  # Case-insensitive

# Starts with / Ends with
Model.filter(name__startswith='A')
Model.filter(name__istartswith='a')  # Case-insensitive
Model.filter(name__endswith='e')
Model.filter(name__iendswith='E')    # Case-insensitive

# Numeric comparisons
Model.filter(age__gt=18)       # Greater than
Model.filter(age__gte=18)      # Greater than or equal
Model.filter(age__lt=65)       # Less than
Model.filter(age__lte=65)      # Less than or equal

# Range
Model.filter(age__range=[18, 65])

# In list
Model.filter(status__in=['active', 'pending'])

# Null checks
Model.filter(name__isnull=True)
Model.filter(name__not_isnull=True)

# Date/time lookups
Model.filter(created_at__year=2023)
Model.filter(created_at__month=12)
Model.filter(created_at__day=25)
Model.filter(created_at__date='2023-12-25')

# Related field lookups
Model.filter(author__name='Alice')
Model.filter(author__posts__count__gt=5)

Bulk Operations

Efficient operations for handling multiple records.

class BulkCreateQuery:
    """Bulk create query for inserting multiple records."""
    
    async def bulk_create(self, objects, batch_size=None, ignore_conflicts=False):
        """
        Create multiple model instances efficiently.
        
        Args:
            objects (list): List of model instances to create
            batch_size (int, optional): Number of objects per batch
            ignore_conflicts (bool): Ignore constraint conflicts
            
        Returns:
            list: Created model instances
        """

class BulkUpdateQuery:
    """Bulk update query for updating multiple records."""
    
    async def bulk_update(self, objects, fields, batch_size=None):
        """
        Update multiple model instances efficiently.
        
        Args:
            objects (list): List of model instances to update
            fields (list): Fields to update
            batch_size (int, optional): Number of objects per batch
            
        Returns:
            int: Number of updated records
        """

# Usage
users = [User(name=f'User {i}') for i in range(100)]
await User.bulk_create(users)

# Bulk update
users = await User.all()
for user in users:
    user.status = 'active'
await User.bulk_update(users, ['status'])

Raw SQL Queries

Execute raw SQL queries when ORM queries are insufficient.

class RawSQLQuery:
    """Raw SQL query executor."""
    
    @classmethod
    async def raw(cls, sql, values=None):
        """
        Execute raw SQL query.
        
        Args:
            sql (str): SQL query string
            values (list, optional): Query parameters
            
        Returns:
            list: Query results
        """

# Usage
results = await User.raw('SELECT * FROM users WHERE age > ?', [18])

Aggregation

Aggregate functions for computing values across multiple records.

# Count
count = await User.all().count()
active_count = await User.filter(is_active=True).count()

# Aggregate with annotations
from tortoise.functions import Count, Sum, Avg, Max, Min

# Annotate with aggregates
users_with_post_count = await User.annotate(
    post_count=Count('posts')
).all()

# Aggregate values
stats = await Post.aggregate(
    total_posts=Count('id'),
    avg_views=Avg('view_count'),
    max_views=Max('view_count'),
    min_views=Min('view_count')
)

Usage Examples

Basic Filtering

from tortoise.models import Model
from tortoise import fields

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    email = fields.CharField(max_length=100)
    age = fields.IntField()
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)

# Simple filters
active_users = await User.filter(is_active=True).all()
adult_users = await User.filter(age__gte=18).all()
alice_users = await User.filter(name__icontains='alice').all()

# Chaining filters
recent_adult_users = await User.filter(
    age__gte=18,
    created_at__gte=datetime(2023, 1, 1)
).order_by('-created_at').limit(10).all()

Complex Queries with Q

from tortoise.queryset import Q

# OR conditions
young_or_senior = await User.filter(
    Q(age__lt=25) | Q(age__gt=65)
).all()

# AND with NOT
active_non_admin = await User.filter(
    Q(is_active=True) & ~Q(name__in=['admin', 'root'])
).all()

# Complex nested conditions
complex_query = await User.filter(
    (Q(name__startswith='A') | Q(name__startswith='B')) &
    Q(age__gte=18) &
    ~Q(email__endswith='.temp')
).all()

Relationships and Joins

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    author = fields.ForeignKeyField('models.User', related_name='posts')
    tags = fields.ManyToManyField('models.Tag', related_name='posts')

# Select related (JOIN)
posts_with_authors = await Post.select_related('author').all()
for post in posts_with_authors:
    print(f"{post.title} by {post.author.name}")  # No additional query

# Prefetch related (separate queries)
posts_with_tags = await Post.prefetch_related('tags').all()
for post in posts_with_tags:
    for tag in post.tags:  # No additional queries
        print(tag.name)

# Filter by related fields
python_posts = await Post.filter(tags__name='python').all()
alice_posts = await Post.filter(author__name='Alice').all()

Aggregation Examples

from tortoise.functions import Count, Sum, Avg

# Simple aggregation
total_users = await User.all().count()
active_users = await User.filter(is_active=True).count()

# Annotations
users_with_post_count = await User.annotate(
    post_count=Count('posts')
).filter(post_count__gt=5).all()

# Group by with aggregation
tag_stats = await Tag.annotate(
    post_count=Count('posts')
).filter(post_count__gt=0).order_by('-post_count').all()

# Multiple aggregates
post_stats = await Post.aggregate(
    total_posts=Count('id'),
    unique_authors=Count('author', distinct=True)
)

Bulk Operations

# Bulk create
users_data = [
    {'name': f'User {i}', 'email': f'user{i}@example.com', 'age': 20 + i}
    for i in range(1000)
]
users = [User(**data) for data in users_data]
created_users = await User.bulk_create(users, batch_size=100)

# Bulk update
await User.filter(is_active=False).update(is_active=True)

# Bulk delete
await User.filter(created_at__lt=datetime(2020, 1, 1)).delete()

Pagination

# Manual pagination
page_size = 20
page_number = 1

users = await User.filter(is_active=True).order_by('name').offset(
    (page_number - 1) * page_size
).limit(page_size).all()

total_count = await User.filter(is_active=True).count()
has_next = total_count > page_number * page_size

Raw SQL

# Raw query returning model instances
users = await User.raw(
    'SELECT * FROM users WHERE age > ? AND name LIKE ?',
    [18, 'A%']
)

# Raw query with connection
from tortoise import connections
conn = connections.get('default')
result = await conn.execute_query(
    'SELECT COUNT(*) as count FROM users WHERE is_active = ?',
    [True]
)

Install with Tessl CLI