API Reference

This page provides detailed documentation for all LightODM classes and functions.

Core Classes

MongoBaseModel

class lightodm.MongoBaseModel(*, _id: str | None = <factory>, **extra_data: Any)[source]

Bases: BaseModel

Base class for MongoDB document models with ODM functionality.

Provides both synchronous and asynchronous methods for CRUD operations. Maps Pydantic ‘id’ field to MongoDB ‘_id’ field.

Subclasses must define an inner Settings class with ‘name’ attribute:

Example

class User(MongoBaseModel):

class Settings:: name = “users”

name: str email: str age: Optional[int] = None

# Sync usage user = User(name=”John”, email=”john@example.com”) user.save()

found_user = User.get(“some_id”) users = User.find({“age”: {“$gt”: 18}})

# Async usage await user.asave() found_user = await User.aget(“some_id”) users = await User.afind({“age”: {“$gt”: 18}})

Base class for MongoDB document models with ODM functionality.

Settings Class

Every model must define an inner Settings class with the following attribute:

Settings.name: str: MongoDB collection name for this model.

Field Mapping

id: str | None: Document ID that maps to MongoDB _id field. Auto-generated using ObjectId if not provided.

Synchronous CRUD Methods

save(exclude_none: bool = False) → str[source]

Save/upsert the document (synchronous).

Parameters:: exclude_none – If True, exclude fields with None values from update
Returns:: Document ID

delete() → bool[source]

Delete the document (synchronous).

Returns:: True if document was deleted, False otherwise

classmethod get(id: str) → T | None[source]

Retrieve a document by ID (synchronous).

Parameters:: id – Document ID
Returns:: Model instance or None if not found

classmethod find_one(filter: dict, **kwargs) → T | None[source]

Find a single document (synchronous).

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find_one (e.g., sort, projection)

Returns:

Model instance or None if not found

classmethod find(filter: dict, **kwargs) → List[T][source]

Find multiple documents (synchronous).

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find (e.g., sort, limit, skip, projection)

Returns:

List of model instances

classmethod find_iter(filter: dict, **kwargs) → Iterator[T][source]

Find multiple documents with iterator (synchronous). Useful for large result sets to avoid loading all into memory.

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find

Yields:

Model instances one at a time

classmethod count(filter: dict = None) → int[source]

Count documents matching filter (synchronous).

Parameters:: filter – MongoDB filter dictionary (default: {} for all documents)
Returns:: Number of matching documents

classmethod update_one(filter: dict, update: dict, upsert: bool = False) → bool[source]

Update a single document (synchronous).

Parameters:

filter – MongoDB filter dictionary
update – MongoDB update dictionary (should include operators like $set)
upsert – If True, insert document if not found

Returns:

True if document was modified, False otherwise

classmethod update_many(filter: dict, update: dict) → int[source]

Update multiple documents (synchronous).

Parameters:

filter – MongoDB filter dictionary
update – MongoDB update dictionary (should include operators like $set)

Returns:

Number of documents modified

classmethod delete_one(filter: dict) → bool[source]

Delete a single document (synchronous).

Parameters:: filter – MongoDB filter dictionary
Returns:: True if document was deleted, False otherwise

classmethod delete_many(filter: dict) → int[source]

Delete multiple documents (synchronous).

Parameters:: filter – MongoDB filter dictionary
Returns:: Number of documents deleted

classmethod aggregate(pipeline: List[dict], **kwargs) → List[dict][source]

Run aggregation pipeline (synchronous).

Parameters:

pipeline – MongoDB aggregation pipeline
**kwargs – Additional arguments passed to aggregate

Returns:

List of result documents

classmethod insert_many(documents: List[T]) → List[str][source]

Insert multiple documents (synchronous).

Parameters:: documents – List of model instances
Returns:: List of inserted document IDs

Asynchronous CRUD Methods

async asave(exclude_none: bool = False) → str[source]

Save/upsert the document (asynchronous).

Parameters:: exclude_none – If True, exclude fields with None values from update
Returns:: Document ID

async adelete() → bool[source]

Delete the document (asynchronous).

Returns:: True if document was deleted, False otherwise

async classmethod aget(id: str) → T | None[source]

Retrieve a document by ID (asynchronous).

Parameters:: id – Document ID
Returns:: Model instance or None if not found

async classmethod afind_one(filter: dict, **kwargs) → T | None[source]

Find a single document (asynchronous).

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find_one (e.g., sort, projection)

Returns:

Model instance or None if not found

async classmethod afind(filter: dict, **kwargs) → List[T][source]

Find multiple documents (asynchronous).

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find (e.g., sort, limit, skip, projection)

Returns:

List of model instances

classmethod afind_iter(filter: dict, **kwargs) → AsyncIterator[T][source]

Find multiple documents with async iterator. Useful for large result sets to avoid loading all into memory.

Parameters:

filter – MongoDB filter dictionary
**kwargs – Additional arguments passed to find

Yields:

Model instances one at a time

async classmethod acount(filter: dict = None) → int[source]

Count documents matching filter (asynchronous).

Parameters:: filter – MongoDB filter dictionary (default: {} for all documents)
Returns:: Number of matching documents

async classmethod aupdate_one(filter: dict, update: dict, upsert: bool = False) → bool[source]

Update a single document (asynchronous).

Parameters:

filter – MongoDB filter dictionary
update – MongoDB update dictionary (should include operators like $set)
upsert – If True, insert document if not found

Returns:

True if document was modified, False otherwise

async classmethod aupdate_many(filter: dict, update: dict) → int[source]

Update multiple documents (asynchronous).

Parameters:

filter – MongoDB filter dictionary
update – MongoDB update dictionary (should include operators like $set)

Returns:

Number of documents modified

async classmethod adelete_one(filter: dict) → bool[source]

Delete a single document (asynchronous).

Parameters:: filter – MongoDB filter dictionary
Returns:: True if document was deleted, False otherwise

async classmethod adelete_many(filter: dict) → int[source]

Delete multiple documents (asynchronous).

Parameters:: filter – MongoDB filter dictionary
Returns:: Number of documents deleted

async classmethod aaggregate(pipeline: List[dict], **kwargs) → List[dict][source]

Run aggregation pipeline (asynchronous).

Parameters:

pipeline – MongoDB aggregation pipeline
**kwargs – Additional arguments passed to aggregate

Returns:

List of result documents

async classmethod ainsert_many(documents: List[T]) → List[str][source]

Insert multiple documents (asynchronous).

Parameters:: documents – List of model instances
Returns:: List of inserted document IDs

Collection Access Methods

classmethod get_collection() → Collection[source]

Get synchronous MongoDB collection.

Override this method to provide custom connection logic.

Returns:: PyMongo Collection instance

async classmethod get_async_collection() → AsyncIOMotorCollection[source]

Get asynchronous MongoDB collection.

Override this method to provide custom connection logic.

Returns:: Motor AsyncIOMotorCollection instance

Internal Methods

_to_mongo_dict(exclude_none: bool = False) → dict[source]

Convert model to dictionary for MongoDB, handling id -> _id mapping.

Only serializes Pydantic fields - class attributes like collection_name are automatically excluded.

Parameters:: exclude_none – If True, exclude fields with None values
Returns:: Dictionary suitable for MongoDB insertion/update

classmethod _from_mongo_dict(data: dict) → T | None[source]

Create model instance from MongoDB document.

Parameters:: data – MongoDB document dictionary
Returns:: Model instance or None if data is None

classmethod _get_collection_name() → str[source]: Get the collection name from Settings.name

classmethod _validate_collection_name()[source]: Ensure Settings.name is defined in subclass

Connection Management

MongoConnection

class lightodm.MongoConnection[source]

Bases: object

Singleton MongoDB connection manager supporting both sync (pymongo) and async (motor) clients with thread-safety.

The connection is configured via environment variables: - MONGO_URL: MongoDB connection URL - MONGO_USER: MongoDB username - MONGO_PASSWORD: MongoDB password - MONGO_DB_NAME: Database name

Example

# Sync usage conn = MongoConnection() db = conn.database collection = conn.get_collection(“users”)

# Async usage conn = MongoConnection() client = await conn.get_async_client() db = await conn.get_async_database()

Singleton MongoDB connection manager supporting both sync and async clients.

Properties

property client: MongoClient: Get the synchronous MongoDB client

property database: Database: Get the synchronous MongoDB database

Methods

get_collection(collection_name: str) → Collection[source]

Get a synchronous collection.

Parameters:: collection_name – Name of the collection
Returns:: PyMongo Collection instance

async get_async_client() → AsyncIOMotorClient[source]

Get or create the AsyncIOMotorClient and verify connectivity asynchronously.

Returns:: Motor AsyncIOMotorClient instance

async get_async_database(db_name: str | None = None) → AsyncIOMotorDatabase[source]

Get the asynchronous MongoDB database.

Parameters:: db_name – Optional database name override
Returns:: Motor AsyncIOMotorDatabase instance

close_connection()[source]: Close both sync and async clients if present.

Helper Functions

connect()

lightodm.connect(url: str | None = None, username: str | None = None, password: str | None = None, db_name: str | None = None) → Database[source]

Initialize MongoDB connection with optional explicit parameters.

If parameters are not provided, they will be read from environment variables:

MONGO_URL
MONGO_USER
MONGO_PASSWORD
MONGO_DB_NAME

Parameters:

url – MongoDB connection URL (optional)
username – MongoDB username (optional)
password – MongoDB password (optional)
db_name – Database name (optional)

Returns:

PyMongo Database instance

Example:

# Connect with explicit parameters
db = connect(
    url="mongodb://localhost:27017",
    username="myuser",
    password="mypass",
    db_name="mydb"
)

# Or use environment variables
db = connect()

Convenience function to initialize MongoDB connection with the singleton pattern.

Parameters:

url (str, optional) – MongoDB connection URL (e.g., mongodb://localhost:27017)
username (str, optional) – MongoDB username
password (str, optional) – MongoDB password
db_name (str, optional) – Default database name

Returns:

MongoDB database instance

Return type:

pymongo.database.Database

Environment Variables

If parameters are not provided, the function will attempt to read from environment variables:

MONGO_URL: MongoDB connection URL
MONGO_USER: MongoDB username
MONGO_PASSWORD: MongoDB password
MONGO_DB_NAME: Default database name

Example

from lightodm import connect, MongoBaseModel

# Connect explicitly
connect(
    url="mongodb://localhost:27017",
    username="myuser",
    password="mypass",
    db_name="mydb"
)

# Or using environment variables
# MONGO_URL=mongodb://localhost:27017
# MONGO_USER=myuser
# MONGO_PASSWORD=mypass
# MONGO_DB_NAME=mydb
connect()

generate_id()

lightodm.generate_id() → str[source]

Generate a new MongoDB ObjectId as a string.

Returns:: String representation of a new ObjectId

Generate a unique ID for MongoDB documents.

Parameters:: \**kwargs (dict) – Optional keyword arguments to generate deterministic hash-based ID
Returns:: Unique ID string (ObjectId or MD5 hash)
Return type:: str

Behavior

If called without arguments: Returns a new ObjectId string
If called with keyword arguments: Returns MD5 hash of sorted key-value pairs (deterministic)

Example

from lightodm import generate_id

# Random ObjectId
id1 = generate_id()  # '507f1f77bcf86cd799439011'

# Deterministic hash-based ID
id2 = generate_id(user_id="123", type="profile")
id3 = generate_id(type="profile", user_id="123")
assert id2 == id3  # Same hash for same inputs

Type Definitions

Collection Types

LightODM uses the following collection types from PyMongo and Motor:

class pymongo.collection.Collection[source]: Synchronous MongoDB collection (from PyMongo).

class motor.motor_asyncio.AsyncIOMotorCollection: Asynchronous MongoDB collection (from Motor).

TypeVar

lightodm.T: TypeVar

Type variable bound to MongoBaseModel for generic class methods.

Used internally for type hinting to ensure methods return instances of the correct model class.

Configuration

Model Configuration

LightODM models use Pydantic v2’s model_config with the following settings:

from pydantic import ConfigDict

model_config = ConfigDict(
    populate_by_name=True,  # Allow field population by alias
    extra='allow'           # Allow extra fields beyond model definition
)

populate_by_name: Enables using either the field name or alias (_id / id) when creating models.
extra=’allow’: Allows storing arbitrary extra fields on the model. Extra fields are preserved when saving to MongoDB.

Environment Variables

The following environment variables are recognized by MongoConnection and connect():

MONGO_URL

MongoDB connection URL.

Example: mongodb://localhost:27017

MONGO_USER: MongoDB username for authentication.

MONGO_PASSWORD: MongoDB password for authentication.

MONGO_DB_NAME: Default database name to use.

Exceptions

LightODM raises the following exceptions:

NotImplementedError

Raised when:

A model doesn’t define a Settings class
Settings.name is not defined
get_collection() or get_async_collection() is not implemented

ValueError

Raised when:

Attempting to save a document without an ID
MongoDB connection parameters are missing

PyMongo Exceptions

LightODM also propagates exceptions from PyMongo and Motor. Common ones include:

pymongo.errors.ConnectionFailure: Cannot connect to MongoDB
pymongo.errors.OperationFailure: MongoDB operation failed
pymongo.errors.DuplicateKeyError: Unique constraint violation

See PyMongo documentation for complete exception reference.

Usage Patterns

Bring Your Own Connection (BYOC)

For advanced use cases, override collection methods to use custom connection logic:

from lightodm import MongoBaseModel
from pymongo import MongoClient
from motor.motor_asyncio import AsyncIOMotorClient

# Global connection instances
sync_client = MongoClient("mongodb://localhost:27017")
async_client = AsyncIOMotorClient("mongodb://localhost:27017")
db_name = "myapp"

class User(MongoBaseModel):
    class Settings:
        name = "users"

    name: str
    email: str

    @classmethod
    def get_collection(cls):
        return sync_client[db_name][cls.Settings.name]

    @classmethod
    async def get_async_collection(cls):
        return async_client[db_name][cls.Settings.name]

Thread Safety

MongoConnection is thread-safe and uses a singleton pattern with locking:

from concurrent.futures import ThreadPoolExecutor
from lightodm import connect, MongoBaseModel

connect()  # Initialize once

class User(MongoBaseModel):
    class Settings:
        name = "users"

    name: str

def save_user(i):
    user = User(name=f"User {i}")
    user.save()

# Safe to use from multiple threads
with ThreadPoolExecutor(max_workers=10) as executor:
    executor.map(save_user, range(100))

Async Context

Always use async methods in async contexts:

import asyncio
from lightodm import MongoBaseModel

class User(MongoBaseModel):
    class Settings:
        name = "users"

    name: str

async def main():
    # Use async methods
    user = User(name="Async User")
    await user.asave()

    users = await User.afind({})

    async for user in User.afind_iter({}):
        print(user.name)

asyncio.run(main())