### Install Motor-ODM using pip

Source: https://github.com/codello/motor-odm/blob/master/README.md

This command installs the Motor-ODM library, enabling its use in your Python projects for interacting with MongoDB.

```shell
pip install Motor-ODM
```

--------------------------------

### Quick Start: User Model and Database Operations with Motor-ODM

Source: https://github.com/codello/motor-odm/blob/master/README.md

This Python snippet demonstrates the basic usage of Motor-ODM. It shows how to define a custom document model 'User' inheriting from 'Document', configure its MongoDB collection, connect to a database using AsyncIOMotorClient, create a new document instance, save it to the database asynchronously, and perform a query to retrieve all documents of that type.

```python
from motor.motor_asyncio import AsyncIOMotorClient
from motor_odm import Document

# Create a custom model by subclassing Document
class User(Document):
    class Mongo:
        # Set the collection name
        collection = "users"
    
    # Add attributes to your model
    username: str
    age: int

# Connect your model to a database
client = AsyncIOMotorClient(...)
Document.use(client.get_default_database())

# Create documents and save them to the database
u = User(username="John", age=20)
await u.insert()

# Query the database
async for user in User.all():
    print(user.username)
```

--------------------------------

### Create and Insert Documents into MongoDB (Python)

Source: https://context7.com/codello/motor-odm/llms.txt

Demonstrates how to create new document instances and insert them into MongoDB using Motor-ODM. It covers single document insertion with automatic ID assignment and validation, handling duplicate key errors, and batch insertion of multiple documents.

```python
# Create a single document
user = User(username="john_doe", email="john@example.com", age=30)
success = await user.insert()
# success = True, user.id is now set to ObjectId

# Insert returns False on duplicate key errors
duplicate = User(username="john_doe", email="john@example.com", age=30)
duplicate.id = user.id  # Same ID
result = await duplicate.insert()
# result = False (duplicate key)

# Batch insert multiple documents
users = [
    User(username="alice", email="alice@example.com", age=25),
    User(username="bob", email="bob@example.com", age=35),
    User(username="charlie", email="charlie@example.com", age=28)
]
await User.insert_many(*users)
# All users now have their .id field populated
print([user.id for user in users])
# [ObjectId('...'), ObjectId('...'), ObjectId('...')]
```

--------------------------------

### FastAPI Integration with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

This Python code showcases seamless integration of Motor-ODM with FastAPI. It defines a `User` document and sets up CRUD (Create, Read, Update, Delete) endpoints for user management. Motor-ODM documents can be directly used as response models in FastAPI, benefiting from automatic serialization and alias handling.

```python
from fastapi import FastAPI, HTTPException
from motor.motor_asyncio import AsyncIOMotorClient
from motor_odm import Document, q
from typing import List

app = FastAPI()

class User(Document):
    class Mongo:
        collection = "users"

    username: str
    email: str
    age: int

@app.on_event("startup")
async def startup():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    Document.use(client["myapp"])

# Motor-ODM documents work directly as response models
@app.post("/users", response_model=User)
async def create_user(user: User):
    await user.insert()
    return user  # Serializes with proper alias handling (_id)

@app.get("/users/{username}", response_model=User)
async def get_user(username: str):
    user = await User.find_one(q(username=username))
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users", response_model=List[User])
async def list_users(min_age: int = 0):
    users = [user async for user in User.find(q(age__gte=min_age))]
    return users

@app.put("/users/{username}", response_model=User)
async def update_user(username: str, age: int):
    user = await User.find_one_and_update(
        q(username=username),
        {"$set": {"age": age}},
        return_document=ReturnDocument.AFTER
    )
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.delete("/users/{username}")
async def delete_user(username: str):
    user = await User.find_one_and_delete(q(username=username))
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"deleted": username}
```

--------------------------------

### Document Inheritance and Abstract Models in Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

Explains how to implement document inheritance and abstract base models in Motor-ODM. It shows creating a base document with common fields like timestamps and then inheriting from it to create concrete document models, including defining collection names and indexes.

```python
from motor_odm import Document, Field
from pymongo import IndexModel
from datetime import datetime
from typing import Optional

# Abstract base document (no collection required)
# class TimestampedDocument(Document, abstract=True):
#     created_at: datetime = Field(default_factory=datetime.utcnow)
#     updated_at: datetime = Field(default_factory=datetime.utcnow)

# Inherit from abstract base
# class User(TimestampedDocument):
#     class Mongo:
#         collection = "users"
#         indexes = [
#             IndexModel("email", unique=True),
#             IndexModel([("username", 1)])
#         ]
# 
#     username: str
#     email: str

# class Product(TimestampedDocument):
#     class Mongo:
#         collection = "products"
#         indexes = [IndexModel("sku", unique=True)]
# 
#     sku: str
#     name: str
#     price: float

# Both User and Product have created_at and updated_at fields
# user = User(username="alice", email="alice@example.com")
# await user.insert()
# print(user.created_at)  # datetime object

# Initialize indexes (creates/updates indexes in MongoDB)
# await User.init_indexes(drop=True)  # Drops changed indexes
# await Product.init_indexes(drop=False)  # Only creates missing indexes
```

--------------------------------

### Query Documents from MongoDB (Python)

Source: https://context7.com/codello/motor-odm/llms.txt

Shows how to retrieve documents from MongoDB using Motor-ODM. It includes methods for finding all documents, finding documents with filter dictionaries, finding a single document, querying by document ID, and using the `q()` query builder for complex queries with logical operators.

```python
from motor_odm import q

# Find all documents
all_users = [user async for user in User.find()]

# Find with filter dictionary
admins = [user async for user in User.find({"is_admin": True})]

# Find one document
john = await User.find_one({"username": "john_doe"})
print(john.email)  # "john@example.com"

# Query by ID
user_by_id = await User.find_one({"_id": some_object_id})

# Using q() query builder - simple equality
 young_user = await User.find_one(q(age=25))

# Using q() with operators (suffix notation)
adults = [user async for user in User.find(q(age__gte=18))]

# Multiple operators on same field
users_25_to_40 = [user async for user in User.find(q(age__gte=25, age__lte=40))]
# Generates: {"age": {"$gte": 25, "$lte": 40}}

# Combining queries with logical operators
query = (q(is_admin=True) & q(age__gt=30)) | q(username="special_user")
results = [user async for user in User.find(query)]
# Generates: {"$or": [{"$
and": [{"is_admin": True}, {"age": {"$gt": 30}}]}, {"username": "special_user"}]}

# Count documents
total_users = await User.count_documents()
admin_count = await User.count_documents({"is_admin": True})
```

--------------------------------

### Define Document Model and Connect to MongoDB (Python)

Source: https://context7.com/codello/motor-odm/llms.txt

Defines a MongoDB document model `User` using Motor-ODM and Pydantic, and sets up the connection to MongoDB using Motor. It shows how to specify the collection name and use Pydantic field definitions for document attributes. The database connection is registered globally or per document class.

```python
from motor.motor_asyncio import AsyncIOMotorClient
from motor_odm import Document
from pydantic import Field
from typing import Optional

# Define a document model
class User(Document):
    class Mongo:
        collection = "users"  # Required collection name

    # Use Pydantic field definitions
    username: str
    email: str
    age: int
    is_admin: bool = False

# Connect to MongoDB and register database
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client["myapp"]
Document.use(db)  # All document classes will use this database

# Alternative: Set database per document class
User.use(db)
```

--------------------------------

### Advanced Query Building with Motor-ODM Query Builder

Source: https://context7.com/codello/motor-odm/llms.txt

Illustrates advanced usage of the Motor-ODM query builder, including querying by single or multiple IDs, handling `None` IDs, using operator suffixes (like `__ne`, `__gt`) for MongoDB operators, converting snake_case to camelCase for operators, and utilizing special query types like schema, text, and where clauses.

```python
from motor_odm import q, Query

# Query by single ID
# user = await User.find_one(q(some_object_id))
# Generates: {"_id": some_object_id}

# Query by multiple IDs
# users = [user async for user in User.find(q(id1, id2, id3))]
# Generates: {"_id": {"$in": [id1, id2, id3]}}

# None ID creates unmatchable query
# empty_query = q(None)
# Generates: {"X": {"$in": []}}

# Operator suffix transforms to MongoDB operators
# query = q(
#     username__ne="admin",
#     age__gt=18,
#     age__lt=65,
#     email__regex="@company.com$"
# )
# Generates: {
#     "username": {"$ne": "admin"},
#     "age": {"$gt": 18, "$lt": 65},
#     "email": {"$regex": "@company.com$"}
# }

# CamelCase operators (snake_case converts to camelCase)
# query = q(flags__bits_all_set=7, tags__all=["python", "mongodb"])
# Generates: {"flags": {"$bitsAllSet": 7}, "tags": {"$all": ["python", "mongodb"]}}

# Special query types
# schema_query = Query.schema({"properties": {"age": {"minimum": 18}}})
# text_query = Query.text("search term", language="en", case_sensitive=False)
# expr_query = Query.expr({"$gt": ["$field1", "$field2"]})
# where_query = Query.where("this.age > 18")

# Add comment to query for debugging
# query = q(age__gt=18).comment("Find adults for analytics")

# Extend existing queries
# base_query = q(is_admin=True)
# base_query.extend(age__gt=25)
# base_query now: {"is_admin": True, "age": {"$gt": 25}}
```

--------------------------------

### MongoDB Collection Options with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

This code demonstrates how to configure MongoDB collection options such as read preferences, write concerns, and read concerns directly within the Motor-ODM `Mongo` class. This allows for centralized and declarative configuration of how your application interacts with MongoDB collections.

```python
from motor_odm import Document
from pymongo import ReadPreference, WriteConcern
from pymongo.read_concern import ReadConcern
from bson.codec_options import CodecOptions

class CriticalData(Document):
    class Mongo:
        collection = "critical_data"
        codec_options = CodecOptions(tz_aware=True)
        read_preference = ReadPreference.PRIMARY
        write_concern = WriteConcern(w="majority", j=True)
        read_concern = ReadConcern(level="majority")

    value: str
    sensitive: bool = True

# Collection automatically uses configured options
data = CriticalData(value="important")
await data.insert()  # Uses write_concern from Mongo class

# Access the configured collection
collection = CriticalData.collection()
print(collection.write_concern)  # WriteConcern(w='majority', j=True)
```

--------------------------------

### Atomic Find and Replace in MongoDB with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

Demonstrates how to atomically find a document by a query and replace it with a new document using Motor-ODM. It specifies the `return_document` option to retrieve the document after the replacement.

```python
from motor_odm import q, ReturnDocument

# Assume User model is defined elsewhere
# replacement = User(username="john_doe", email="updated@example.com", age=33)
# user = await User.find_one_and_replace(
#     q(username="john_doe"),
#     replacement,
#     return_document=ReturnDocument.AFTER
# )
# user now contains the replaced document
```

--------------------------------

### Initialize Missing Indexes with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

This snippet demonstrates how to initialize missing indexes for a Motor-ODM document within a session. It ensures that only absent indexes are created, preserving any pre-existing ones. It's designed to be used within a transactional session.

```python
async with await client.start_session() as session:
    await Article.init_indexes(session=session)
```

--------------------------------

### Index Management in Motor-ODM for MongoDB

Source: https://context7.com/codello/motor-odm/llms.txt

Details how to define and manage MongoDB indexes using Motor-ODM's `Mongo` inner class configuration. It covers simple field indexes, compound indexes, text indexes, and indexes with specific options like `sparse` and `partialFilterExpression`, as well as methods for initializing indexes.

```python
from motor_odm import Document
from pymongo import IndexModel, ASCENDING, DESCENDING, TEXT
from datetime import datetime
from typing import Optional

# class Article(Document):
#     class Mongo:
#         collection = "articles"
#         indexes = [
#             # Simple field index with unique constraint
#             IndexModel("slug", unique=True),
# 
#             # Compound index
#             IndexModel([("author", ASCENDING), ("published_at", DESCENDING)]),
# 
#             # Text index for full-text search
#             IndexModel([("title", TEXT), ("content", TEXT)]),
# 
#             # Index with options
#             IndexModel(
#                 "email",
#                 unique=True,
#                 sparse=True,
#                 partialFilterExpression={"email": {"$exists": True}}
#             )
#         ]
# 
#     title: str
#     content: str
#     author: str
#     slug: str
#     published_at: datetime
#     email: Optional[str] = None

# Create/update indexes in database
# await Article.init_indexes(drop=True)
# Compares defined indexes with existing ones
# Drops changed indexes and recreates them
# Creates new indexes

# Initialize without dropping (safer for production)
# await Article.init_indexes(drop=False)
```

--------------------------------

### Update and Save Documents in MongoDB (Python)

Source: https://context7.com/codello/motor-odm/llms.txt

Illustrates how to update and save documents in MongoDB using Motor-ODM. It covers updating an existing document and saving it, using `save` for new documents (upsert by default), explicitly controlling upsert behavior, reloading a document from the database, and performing atomic find and update operations.

```python
# Update and save existing document
user = await User.find_one(q(username="john_doe"))
user.age = 31
changed = await user.save()
# changed = True if document was modified

# Save new document (upsert by default)
new_user = User(username="new_user", email="new@example.com", age=22)
await new_user.save()  # Inserts document and sets new_user.id

# Update without upsert
user.email = "newemail@example.com"
result = await user.save(upsert=False)
# result = True if document exists and was updated

# Reload document from database (discard local changes)
user.age = 99  # Local change
await user.reload()
print(user.age)  # Original value from database

# Atomic find and update with MongoDB operators
updated_user = await User.find_one_and_update(
    q(username="john_doe"),
    {"$
set": {"age": 32}, "$
inc": {"login_count": 1}},
    return_document=ReturnDocument.AFTER
)
print(updated_user.age)  # 32
```

--------------------------------

### Python: Async MongoDB Transactional Transfer

Source: https://context7.com/codello/motor-odm/llms.txt

Demonstrates a transactional money transfer between two accounts using Motor-ODM and asynchronous MongoDB sessions. It ensures that both the sender's and receiver's balances are updated atomically, or the entire operation is rolled back. Requires Motor and Pydantic.

```python
from motor.motor_asyncio import AsyncIOMotorClient
from motor_odm import Document, q
from bson.objectid import ObjectId
from pymongo import ReturnDocument

class Account(Document):
    class Mongo:
        collection = "accounts"

    account_id: str
    balance: float

class Transaction(Document):
    class Mongo:
        collection = "transactions"

    from_account: str
    to_account: str
    amount: float

client = AsyncIOMotorClient("mongodb://localhost:27017/?replicaSet=rs0")
db = client["banking"]
Document.use(db)

async def transfer_money(from_id: str, to_id: str, amount: float):
    async with await client.start_session() as session:
        async with session.start_transaction():
            # All operations within transaction use the session
            sender = await Account.find_one_and_update(
                q(account_id=from_id),
                {"$inc": {"balance": -amount}},
                session=session,
                return_document=ReturnDocument.AFTER
            )

            if sender.balance < 0:
                await session.abort_transaction()
                raise ValueError("Insufficient funds")

            receiver = await Account.find_one_and_update(
                q(account_id=to_id),
                {"$inc": {"balance": amount}},
                session=session,
                return_document=ReturnDocument.AFTER
            )

            # Record transaction
            txn = Transaction(
                from_account=from_id,
                to_account=to_id,
                amount=amount
            )
            await txn.insert(session=session)

            # Transaction commits automatically at end of context

# Usage
try:
    await transfer_money("ACC001", "ACC002", 100.0)
    print("Transfer successful")
except ValueError as e:
    print(f"Transfer failed: {e}")

```

--------------------------------

### Deleting Documents Individually and in Batches with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

Shows how to delete documents from MongoDB using Motor-ODM. Includes methods for deleting a single document by fetching it first, atomically finding and deleting a document, and batch deleting multiple documents based on a query.

```python
# Delete single document
# user = await User.find_one(q(username="john_doe"))
# await user.delete()
# Document removed from database, user object unchanged

# Atomic find and delete
# deleted_user = await User.find_one_and_delete(q(username="alice"))
# Returns the deleted document or None if not found

# Batch delete multiple documents
# users_to_delete = [user async for user in User.find(q(is_admin=False))]
# deleted_count = await User.delete_many(*users_to_delete)
# print(f"Deleted {deleted_count} users")

# Verify deletion
# remaining = await User.count_documents()
```

--------------------------------

### Custom Field Types and Encoders with Motor-ODM

Source: https://context7.com/codello/motor-odm/llms.txt

This section explains how to use custom BSON encoders in Motor-ODM to support Python types not natively handled by MongoDB, such as sets and frozensets. These custom encoders allow Python sets to be stored as arrays in MongoDB and correctly retrieved back into Python set objects.

```python
from motor_odm import Document, SetEncoder, FrozensetEncoder
from bson.codec_options import CodecOptions, TypeRegistry
from typing import Set, FrozenSet

class User(Document):
    class Mongo:
        collection = "users"
        codec_options = CodecOptions(
            type_registry=TypeRegistry([
                SetEncoder(),
                FrozensetEncoder()
            ])
        )

    username: str
    tags: Set[str]
    permissions: FrozenSet[str]

# Sets are automatically converted to lists for storage
user = User(
    username="alice",
    tags={"python", "mongodb", "async"},
    permissions=frozenset(["read", "write"])
)
await user.insert()

# Retrieved as sets
retrieved_user = await User.find_one(q(username="alice"))
print(type(retrieved_user.tags))  # <class 'set'>
print(retrieved_user.tags)  # {"python", "mongodb", "async"}

# MongoDB stores as arrays but Pydantic converts back
# Database document: {"tags": ["python", "mongodb", "async"]}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.