Async CRUD with Motor

Insert Operations

from motor.motor_asyncio import AsyncIOMotorClient
from datetime import datetime

client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.mydb

# Insert one
async def create_user(username: str, email: str):
    result = await db.users.insert_one({
        "username": username,
        "email": email,
        "created_at": datetime.utcnow()
    })
    return result.inserted_id

# Insert many
async def create_users(users: list):
    result = await db.users.insert_many(users)
    return result.inserted_ids

Find Operations

# Find one
async def get_user(user_id):
    return await db.users.find_one({"_id": user_id})

# Find with filter
async def get_active_users():
    cursor = db.users.find({"status": "active"})
    return await cursor.to_list(length=100)

# Find with options
async def get_users_paginated(skip: int = 0, limit: int = 10):
    cursor = db.users.find().skip(skip).limit(limit).sort("created_at", -1)
    return await cursor.to_list(length=limit)

# Iterate async
async def process_all_users():
    async for user in db.users.find():
        await process_user(user)

Update Operations

# Update one
async def update_email(user_id, new_email: str):
    result = await db.users.update_one(
        {"_id": user_id},
        {"$set": {"email": new_email, "updated_at": datetime.utcnow()}}
    )
    return result.modified_count

# Update many
async def deactivate_old_users(days: int = 30):
    cutoff = datetime.utcnow() - timedelta(days=days)
    result = await db.users.update_many(
        {"last_login": {"$lt": cutoff}},
        {"$set": {"status": "inactive"}}
    )
    return result.modified_count

# Find and modify
async def increment_login_count(user_id):
    result = await db.users.find_one_and_update(
        {"_id": user_id},
        {"$inc": {"login_count": 1}, "$set": {"last_login": datetime.utcnow()}},
        return_document=True  # Return updated document
    )
    return result

Delete Operations

# Delete one
async def delete_user(user_id):
    result = await db.users.delete_one({"_id": user_id})
    return result.deleted_count > 0

# Delete many
async def cleanup_inactive_users():
    result = await db.users.delete_many({"status": "deleted"})
    return result.deleted_count

Bulk Operations

from pymongo import InsertOne, UpdateOne, DeleteOne

async def bulk_operations():
    operations = [
        InsertOne({"name": "Alice"}),
        UpdateOne({"name": "Bob"}, {"$set": {"age": 25}}),
        DeleteOne({"name": "Charlie"})
    ]

    result = await db.users.bulk_write(operations)
    return {
        "inserted": result.inserted_count,
        "modified": result.modified_count,
        "deleted": result.deleted_count
    }