mongospec 0.2

Async MongoDB ODM with msgspec integration and automatic collection binding

Minimal async MongoDB ODM built for speed and simplicity, featuring automatic collection binding, msgspec integration, and first-class asyncio support.

---

Table of Contents

1. Installation
2. Quick Start
3. Examples
4. Key Features
5. Core Concepts
   • Document Models
   • Connection Management
   • Collection Binding
   • CRUD Operations
   • Indexes
   • Lifecycle Hooks
6. Contributing
7. License

---

Installation

pip install mongospec

Requires Python 3.13+ and a running MongoDB 6.0+ server.

---

Quick Start

import asyncio
from datetime import datetime
from typing import ClassVar, Sequence

import mongojet
import msgspec

import mongospec
from mongospec import MongoDocument
from mongojet import IndexModel


class User(MongoDocument):
    __collection_name__ = "users"
    __indexes__: ClassVar[Sequence[IndexModel]] = [
        IndexModel(keys=[("email", 1)], options={"unique": True})
    ]

    name: str
    email: str
    created_at: datetime = msgspec.field(default_factory=datetime.now)


async def main() -> None:
    client = await mongojet.create_client("mongodb://localhost:27017")
    await mongospec.init(client.get_database("example_db"), document_types=[User])

    user = User(name="Alice", email="[e-mail]")
    await user.insert()
    print("Inserted:", user)

    fetched = await User.find_one({"email": "[e-mail]"})
    print("Fetched:", fetched)

    await fetched.delete()
    await mongospec.close()


if __name__ == "__main__":
    asyncio.run(main())

---

Examples

All other usage examples have been moved to standalone scripts in the examples/ directory. Each file is self-contained and can be executed directly:

Script	What it covers
quick_start.py	End-to-end “hello world”
document_models.py	Defining typed models & indexes
connection_management.py	Initialising the ODM and binding collections
collection_binding.py	Using models immediately after init
index_creation.py	Unique, compound & text indexes
create_documents.py	Single & bulk inserts, conditional insert
read_documents.py	Queries, cursors, projections
update_documents.py	Field updates, atomic & versioned updates
delete_documents.py	Single & batch deletes
count_documents.py	Fast counts & estimated counts
working_with_cursors.py	Batch processing large result sets
batch_operations.py	Bulk insert / update / delete
atomic_updates.py	Optimistic-locking with version field
upsert_operations.py	Upsert via save and update_one
projection_example.py	Field selection for performance

---

Key Features

• Zero-boilerplate models – automatic collection resolution & binding.
• Async first – built on mongojet, fully await-able API.
• Typed & fast – data classes powered by msgspec for ultra-fast (de)serialization.
• Declarative indexes – define indexes right on the model with familiar pymongo/mongojet IndexModels.
• Batteries included – helpers for common CRUD patterns, bulk and atomic operations, cursors, projections, upserts and more.

---

Core Concepts

Document Models

Define your schema by subclassing MongoDocument and adding typed attributes. See examples/document_models.py.

Connection Management

Initialise once with mongospec.init(...), passing a mongojet.Database and the list of models to bind. See examples/connection_management.py.

Collection Binding

After initialisation every model knows its collection and can be used immediately – no manual wiring required. See examples/collection_binding.py.

CRUD Operations

The MongoDocument class (and its mixins) exposes a rich async CRUD API: insert, find, update, delete, count, cursors, bulk helpers, atomic find_one_and_update, upserts, etc. See scripts in examples/ grouped by operation type.

Indexes

Declare indexes in __indexes__ as a Sequence[IndexModel] (unique, compound, text, …). Indexes are created automatically at init time. See examples/index_creation.py.

Automatic Discovery of Document Models

In addition to manually listing document classes when calling mongospec.init(...), you can use the utility function collect_document_types(...) to automatically discover all models in a package:

from mongospec.utils import collect_document_types

document_types = collect_document_types("myapp.db.models")
await mongospec.init(db, document_types=document_types)

This function supports:

• Recursive import of all submodules in the target package
• Filtering by base class (default: MongoDocument)
• Optional exclusion of abstract or re-exported classes
• Regex or callable-based module filtering
• Graceful handling of import errors

Usage Example:

from mongospec.utils import collect_document_types

# Collect all document models in `myapp.db.models` and its submodules
models = collect_document_types(
    "myapp.db.models",
    ignore_abstract=True,
    local_only=True,
    on_error="warn",
)

await mongospec.init(db, document_types=models)

Advanced options include:

• predicate=... to filter only specific model types
• return_map=True to get a {qualified_name: class} dict
• module_filter=".*models.*" to restrict traversal

See the full function signature in mongospec/utils.py.

---

Lifecycle Hooks

MongoDocument provides two hooks that subclasses can override to inject custom logic before write operations:

Hook	Called by	Purpose
__pre_save__(self) -> None	insert(), insert_one(), insert_many(), save()	Mutate instance fields before serialization
__pre_update__(cls, update) -> dict	update_one(), update_many(), update_by_id(), find_one_and_update()	Modify the update document before execution

Example — automatic updated_at:

from datetime import datetime, UTC
from typing import Any

import msgspec

from mongospec import MongoDocument


class Document(MongoDocument):
    created_at: datetime = msgspec.field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = msgspec.field(default_factory=lambda: datetime.now(UTC))

    def __pre_save__(self) -> None:
        self.updated_at = datetime.now(UTC)

    @classmethod
    def __pre_update__(cls, update: dict[str, Any]) -> dict[str, Any]:
        update.setdefault("$set", {}).setdefault("updated_at", datetime.now(UTC))
        return update

Now every insert, save, or update operation automatically keeps updated_at in sync — no caller-side boilerplate needed.

---

Contributing

Contributions, issues and feature requests are welcome.

---

License

MIT