tessl/pypi-odmantic

AsyncIO MongoDB Object Document Mapper for Python using type hints

—

Pending

Overview

Eval results

Files

Queries

Name: tessl/pypi-odmantic
Author: tessl

ODMantic provides a functional query building system using Python operators and functions to construct MongoDB queries and sort expressions.

Capabilities

Comparison Operators

Functions for building comparison queries against model fields.

def eq(field, value):
    """
    Equality comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB equality query
    """

def ne(field, value):
    """
    Not equal comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB not equal query
    """

def gt(field, value):
    """
    Greater than comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB greater than query
    """

def gte(field, value):
    """
    Greater than or equal comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB greater than or equal query
    """

def lt(field, value):
    """
    Less than comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB less than query
    """

def lte(field, value):
    """
    Less than or equal comparison.
    
    Args:
        field: FieldProxy for the model field
        value: Value to compare against
        
    Returns:
        QueryExpression: MongoDB less than or equal query
    """

Set Operators

Functions for set-based queries.

def in_(field, sequence):
    """
    Value in sequence query.
    
    Args:
        field: FieldProxy for the model field
        sequence: Iterable of values to match against
        
    Returns:
        QueryExpression: MongoDB $in query
    """

def not_in(field, sequence):
    """
    Value not in sequence query.
    
    Args:
        field: FieldProxy for the model field
        sequence: Iterable of values to exclude
        
    Returns:
        QueryExpression: MongoDB $nin query
    """

Pattern Matching

Functions for pattern and text matching.

def match(field, pattern):
    """
    Regular expression pattern matching.
    
    Args:
        field: FieldProxy for the model field
        pattern: Regular expression pattern (str or compiled Pattern)
        
    Returns:
        QueryExpression: MongoDB regex query
    """

Logical Operators

Functions for combining multiple query expressions.

def and_(*elements):
    """
    Logical AND of multiple query expressions.
    
    Args:
        *elements: QueryExpression objects to combine with AND
        
    Returns:
        QueryExpression: MongoDB $and query
    """

def or_(*elements):
    """
    Logical OR of multiple query expressions.
    
    Args:
        *elements: QueryExpression objects to combine with OR
        
    Returns:
        QueryExpression: MongoDB $or query
    """

def nor_(*elements):
    """
    Logical NOR of multiple query expressions.
    
    Args:
        *elements: QueryExpression objects to combine with NOR
        
    Returns:
        QueryExpression: MongoDB $nor query
    """

Sort Operations

Functions for building sort expressions.

def asc(field):
    """
    Ascending sort order.
    
    Args:
        field: FieldProxy for the model field
        
    Returns:
        SortExpression: MongoDB ascending sort
    """

def desc(field):
    """
    Descending sort order.
    
    Args:
        field: FieldProxy for the model field
        
    Returns:
        SortExpression: MongoDB descending sort
    """

Query Types

Core query and sort expression types.

class QueryExpression(dict):
    """MongoDB query expression dictionary."""

class SortExpression(dict):
    """MongoDB sort expression dictionary."""

Usage Examples

Basic Comparisons

from odmantic import Model, AIOEngine
from odmantic.query import eq, ne, gt, gte, lt, lte

class Product(Model):
    name: str
    price: float
    category: str
    stock: int

async def comparison_examples(engine: AIOEngine):
    # Equality (can also use == operator)
    expensive = await engine.find(Product, eq(Product.price, 100.0))
    # Or: expensive = await engine.find(Product, Product.price == 100.0)
    
    # Not equal
    not_electronics = await engine.find(Product, ne(Product.category, "electronics"))
    # Or: not_electronics = await engine.find(Product, Product.category != "electronics")
    
    # Greater than
    expensive_items = await engine.find(Product, gt(Product.price, 50.0))
    # Or: expensive_items = await engine.find(Product, Product.price > 50.0)
    
    # Less than or equal
    affordable = await engine.find(Product, lte(Product.price, 25.0))
    # Or: affordable = await engine.find(Product, Product.price <= 25.0)
    
    # Range query
    mid_range = await engine.find(
        Product, 
        gte(Product.price, 20.0),
        lte(Product.price, 50.0)
    )
    # Or: mid_range = await engine.find(Product, Product.price >= 20.0, Product.price <= 50.0)

Set Operations

from odmantic.query import in_, not_in

async def set_examples(engine: AIOEngine):
    # Find products in specific categories
    tech_products = await engine.find(
        Product, 
        in_(Product.category, ["electronics", "computers", "phones"])
    )
    
    # Exclude out-of-stock items
    available_products = await engine.find(
        Product, 
        not_in(Product.stock, [0])
    )
    
    # Find products with specific names
    featured_products = await engine.find(
        Product,
        in_(Product.name, ["iPhone", "MacBook", "iPad"])
    )

Pattern Matching

from odmantic.query import match
import re

async def pattern_examples(engine: AIOEngine):
    # Find products with names starting with "Pro"
    pro_products = await engine.find(
        Product,
        match(Product.name, r"^Pro")
    )
    
    # Case-insensitive search
    apple_products = await engine.find(
        Product,
        match(Product.name, re.compile(r"apple", re.IGNORECASE))
    )
    
    # Find products with email pattern in description (if you had a description field)
    # email_mentions = await engine.find(Product, match(Product.description, r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"))

Logical Operations

from odmantic.query import and_, or_, nor_

async def logical_examples(engine: AIOEngine):
    # AND: expensive electronics
    expensive_electronics = await engine.find(
        Product,
        and_(
            eq(Product.category, "electronics"),
            gt(Product.price, 100.0)
        )
    )
    
    # OR: either cheap or high stock
    bargain_or_available = await engine.find(
        Product,
        or_(
            lt(Product.price, 20.0),
            gt(Product.stock, 100)
        )
    )
    
    # NOR: neither expensive nor out of stock
    reasonable_available = await engine.find(
        Product,
        nor_(
            gt(Product.price, 200.0),
            eq(Product.stock, 0)
        )
    )
    
    # Complex nested logical operations
    complex_query = await engine.find(
        Product,
        and_(
            or_(
                eq(Product.category, "electronics"),
                eq(Product.category, "computers")
            ),
            gt(Product.price, 50.0),
            gt(Product.stock, 0)
        )
    )

Sorting

from odmantic.query import asc, desc

async def sorting_examples(engine: AIOEngine):
    # Sort by price ascending
    cheap_first = await engine.find(Product, sort=asc(Product.price))
    
    # Sort by price descending
    expensive_first = await engine.find(Product, sort=desc(Product.price))
    
    # Multiple sort criteria (tuple of sort expressions)
    sorted_products = await engine.find(
        Product,
        sort=(desc(Product.category), asc(Product.price))
    )
    
    # Sort with query
    electronics_by_price = await engine.find(
        Product,
        eq(Product.category, "electronics"),
        sort=desc(Product.price)
    )

Complex Query Examples

async def complex_examples(engine: AIOEngine):
    # Find available, reasonably priced electronics
    good_electronics = await engine.find(
        Product,
        and_(
            eq(Product.category, "electronics"),
            gte(Product.price, 10.0),
            lte(Product.price, 500.0),
            gt(Product.stock, 0)
        ),
        sort=asc(Product.price)
    )
    
    # Find products that match multiple criteria
    filtered_products = await engine.find(
        Product,
        or_(
            and_(
                eq(Product.category, "books"),
                lt(Product.price, 30.0)
            ),
            and_(
                in_(Product.category, ["electronics", "computers"]),
                gte(Product.stock, 5),
                lte(Product.price, 200.0)
            )
        ),
        sort=(desc(Product.stock), asc(Product.price)),
        limit=20
    )
    
    # Text search with multiple patterns
    search_results = await engine.find(
        Product,
        or_(
            match(Product.name, r"(?i)(pro|premium|deluxe)"),
            and_(
                match(Product.category, r"(?i)electronics"),
                gt(Product.price, 100.0)
            )
        ),
        sort=desc(Product.price)
    )

Query Building Patterns

def build_product_filter(min_price=None, max_price=None, categories=None, in_stock_only=False):
    """Build a dynamic query based on parameters."""
    conditions = []
    
    if min_price is not None:
        conditions.append(gte(Product.price, min_price))
    
    if max_price is not None:
        conditions.append(lte(Product.price, max_price))
    
    if categories:
        conditions.append(in_(Product.category, categories))
    
    if in_stock_only:
        conditions.append(gt(Product.stock, 0))
    
    if len(conditions) == 1:
        return conditions[0]
    elif len(conditions) > 1:
        return and_(*conditions)
    else:
        return {}  # No conditions

async def dynamic_search(engine: AIOEngine):
    # Use the dynamic filter
    query = build_product_filter(
        min_price=20.0,
        max_price=100.0,
        categories=["electronics", "books"],
        in_stock_only=True
    )
    
    products = await engine.find(Product, query, sort=asc(Product.price))

Working with Dates and ObjectIds

from datetime import datetime, timedelta
from odmantic import ObjectId

class Order(Model):
    customer_id: ObjectId
    created_at: datetime
    total: float

async def date_queries(engine: AIOEngine):
    # Find recent orders
    week_ago = datetime.utcnow() - timedelta(days=7)
    recent_orders = await engine.find(
        Order,
        gte(Order.created_at, week_ago)
    )
    
    # Find orders for specific customer
    customer_orders = await engine.find(
        Order,
        eq(Order.customer_id, ObjectId("507f1f77bcf86cd799439011"))
    )
    
    # Find large orders this month
    month_start = datetime.utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    large_orders = await engine.find(
        Order,
        and_(
            gte(Order.created_at, month_start),
            gt(Order.total, 500.0)
        ),
        sort=desc(Order.total)
    )

Install with Tessl CLI