or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

bson-types.mdengines.mdfields.mdindex.mdindexes.mdmodels.mdqueries.mdsessions.md
tile.json

tessl/pypi-odmantic

AsyncIO MongoDB Object Document Mapper for Python using type hints

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/odmantic@1.0.x

To install, run

npx @tessl/cli install tessl/pypi-odmantic@1.0.0

index.mddocs/

ODMantic

ODMantic is a modern async Object Document Mapper (ODM) for MongoDB in Python that leverages type hints and is built on top of Pydantic for model definition and validation. It provides a developer-friendly API for defining MongoDB document models using standard Python types, supports both synchronous and asynchronous operations with AsyncIO frameworks like FastAPI, and offers features including automatic data validation, JSON serialization and schema generation, functional query building using Python operators, and comprehensive type safety.

Package Information

  • Package Name: odmantic
  • Language: Python
  • Installation: pip install odmantic

Core Imports

import odmantic

Common imports for database operations:

from odmantic import AIOEngine, SyncEngine, Model, Field, ObjectId
from odmantic.config import ODMConfigDict

Basic Usage

from odmantic import AIOEngine, Model, Field
from motor.motor_asyncio import AsyncIOMotorClient
import asyncio

# Define a model
class User(Model):
    name: str
    email: str = Field(unique=True)
    age: int = Field(ge=0)

# Setup async engine
client = AsyncIOMotorClient("mongodb://localhost:27017")
engine = AIOEngine(client, database="my_db")

async def main():
    # Create and save a user
    user = User(name="John Doe", email="john@example.com", age=30)
    await engine.save(user)
    
    # Find users
    users = await engine.find(User, User.age >= 18)
    async for user in users:
        print(f"{user.name}: {user.email}")
    
    # Find one user
    user = await engine.find_one(User, User.email == "john@example.com")
    if user:
        print(f"Found: {user.name}")

asyncio.run(main())

Architecture

ODMantic's architecture provides both async and sync database operations:

  • Engines: Database operation handlers (AIOEngine for async, SyncEngine for sync)
  • Models: Document schemas with type validation using Pydantic
  • Fields: MongoDB-specific field definitions with indexing and constraints
  • Sessions: Transaction and session management for atomic operations
  • Query Builder: Functional query construction using Python operators
  • BSON Types: Native MongoDB types like ObjectId, Int64, Decimal128

This design allows for type-safe MongoDB operations while maintaining high performance through motor (async) and pymongo (sync) drivers.

Capabilities

Database Engines

Async and sync MongoDB operation engines providing complete CRUD functionality, query building, session management, and database configuration.

class AIOEngine:
    def __init__(self, client=None, database="test"): ...
    async def save(self, instance, session=None): ...
    async def find(self, model, *queries, sort=None, skip=0, limit=None, session=None): ...
    async def find_one(self, model, *queries, sort=None, session=None): ...
    async def delete(self, instance, session=None): ...
    async def count(self, model, *queries, session=None): ...

class SyncEngine:
    def __init__(self, client=None, database="test"): ...
    def save(self, instance, session=None): ...
    def find(self, model, *queries, sort=None, skip=0, limit=None, session=None): ...
    def find_one(self, model, *queries, sort=None, session=None): ...
    def delete(self, instance, session=None): ...
    def count(self, model, *queries, session=None): ...

Database Engines

Model Definition

Document model classes with Pydantic integration for validation, serialization, and MongoDB-specific features like embedded documents and references.

class Model:
    def model_copy(self, **changes): ...
    def model_update(self, **changes): ...
    def model_dump_doc(self, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False): ...
    @classmethod
    def model_validate_doc(cls, raw_doc): ...

class EmbeddedModel:
    def model_update(self, **changes): ...

Models

Field Definitions

MongoDB-specific field definitions with indexing, validation, and constraint options.

def Field(
    default=...,
    *,
    key_name=None,
    primary_field=False,
    index=False,
    unique=False,
    default_factory=None,
    title=None,
    description=None,
    **kwargs
): ...

Fields

Query Building

Functional query construction using Python operators for MongoDB queries and sorting.

def eq(field, value): ...
def ne(field, value): ...
def gt(field, value): ...
def gte(field, value): ...
def lt(field, value): ...
def lte(field, value): ...
def in_(field, sequence): ...
def not_in(field, sequence): ...
def match(field, pattern): ...
def and_(*elements): ...
def or_(*elements): ...
def nor_(*elements): ...
def asc(field): ...
def desc(field): ...

Queries

Session Management

Session and transaction support for atomic operations and consistency guarantees.

class AIOSession:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

class AIOTransaction:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

class SyncSession:
    def __enter__(self): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

class SyncTransaction:
    def __enter__(self): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

Sessions

BSON Types

MongoDB BSON types including ObjectId, Int64, Decimal128, Binary, and Regex.

class ObjectId:
    def __init__(self, oid=None): ...
    def __str__(self): ...

class WithBsonSerializer:
    def __bson__(self): ...

BSON Types

Index Management

Index definition and management for MongoDB collections.

class Index:
    def __init__(self, *fields, **kwargs): ...

def Reference(*, key_name=None): ...

Indexes

Types

from typing import Optional, Union, List, Dict, Any, Literal, Callable, Iterable
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorClientSession
from pymongo import MongoClient, IndexModel
from pymongo.client_session import ClientSession
from pydantic import ConfigDict
from pydantic_core import PydanticCustomError

# Engine types
AIOSessionType = Union[AsyncIOMotorClientSession, AIOSession, AIOTransaction, None]
SyncSessionType = Union[ClientSession, SyncSession, SyncTransaction, None]

# Configuration type
class ODMConfigDict(ConfigDict):
    collection: str | None
    parse_doc_with_default_factories: bool
    indexes: Callable[[], Iterable[Index | IndexModel]] | None

# Field proxy for query building
class FieldProxy:
    def __eq__(self, other): ...
    def __ne__(self, other): ...
    def __gt__(self, other): ...
    def __ge__(self, other): ...
    def __lt__(self, other): ...
    def __le__(self, other): ...

# Query and sort expressions
class QueryExpression(Dict[str, Any]): ...
class SortExpression(Dict[str, Literal[-1, 1]]): ...

# Cursors for result iteration
class AIOCursor:
    def __aiter__(self): ...
    def __await__(self): ...

class SyncCursor:
    def __iter__(self): ...

Exceptions

class BaseEngineException(Exception): ...
class DocumentNotFoundError(BaseEngineException): ...
class DuplicateKeyError(BaseEngineException): ...
class DocumentParsingError(ValueError): ...

def ODManticCustomError(error_type: str, message_template: str, context: Union[Dict[str, Any], None] = None) -> PydanticCustomError: ...
def KeyNotFoundInDocumentError(key_name: str) -> PydanticCustomError: ...
def ReferencedDocumentNotFoundError(foreign_key_name: str) -> PydanticCustomError: ...
def IncorrectGenericEmbeddedModelValue(value: Any) -> PydanticCustomError: ...