160 lines
4.7 KiB
Python
160 lines
4.7 KiB
Python
"""
|
|
MongoDB Operations Examples
|
|
|
|
This module provides practical examples of using MongoDB operations through our mixins.
|
|
Each example demonstrates different aspects of CRUD operations and aggregation.
|
|
"""
|
|
|
|
import arrow
|
|
from datetime import datetime
|
|
|
|
from Services.MongoDb.database import MongoDBHandler
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def insert_examples(db) -> None:
|
|
"""Examples of insert operations."""
|
|
# Get the collection
|
|
users_collection = db.get_collection("users")
|
|
products_collection = db.get_collection("products")
|
|
|
|
# Single document insert
|
|
user_doc = {
|
|
"username": "john_doe",
|
|
"email": "john@example.com",
|
|
"age": 30,
|
|
"created_at": datetime.now(),
|
|
}
|
|
result = users_collection.insert_one(user_doc)
|
|
print(f"Inserted user with ID: {result.inserted_id}")
|
|
|
|
# Multiple documents insert
|
|
products = [
|
|
{"name": "Laptop", "price": 999.99, "stock": 50},
|
|
{"name": "Mouse", "price": 29.99, "stock": 100},
|
|
{"name": "Keyboard", "price": 59.99, "stock": 75},
|
|
]
|
|
result = products_collection.insert_many(products)
|
|
print(f"Inserted {len(result.inserted_ids)} products")
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def find_examples(db) -> None:
|
|
"""Examples of find operations."""
|
|
# Get the collections
|
|
users_collection = db.get_collection("users")
|
|
products_collection = db.get_collection("products")
|
|
|
|
# Find one document
|
|
user = users_collection.find_one({"email": "john@example.com"})
|
|
print(f"Found user: {user}")
|
|
|
|
# Find many documents
|
|
products_cursor = products_collection.find({"price": {"$lt": 100}})
|
|
products = list(products_cursor)
|
|
print(f"Found {len(products)} products under $100")
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def update_examples(db) -> None:
|
|
"""Examples of update operations."""
|
|
# Get the collections
|
|
products_collection = db.get_collection("products")
|
|
|
|
# Update single document
|
|
result = products_collection.update_one(
|
|
{"name": "Laptop"}, {"$set": {"price": 899.99, "stock": 45}}
|
|
)
|
|
print(f"Updated {result.modified_count} laptop(s)")
|
|
|
|
# Update multiple documents
|
|
result = products_collection.update_many(
|
|
{"stock": {"$lt": 10}}, {"$set": {"status": "low_stock"}}
|
|
)
|
|
print(f"Updated {result.modified_count} low stock products")
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def delete_examples(db) -> None:
|
|
"""Examples of delete operations."""
|
|
# Get the collections
|
|
users_collection = db.get_collection("users")
|
|
products_collection = db.get_collection("products")
|
|
|
|
# Delete single document
|
|
result = users_collection.delete_one({"email": "john@example.com"})
|
|
print(f"Deleted {result.deleted_count} user")
|
|
|
|
# Delete multiple documents
|
|
result = products_collection.delete_many({"stock": 0})
|
|
print(f"Deleted {result.deleted_count} out-of-stock products")
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def aggregate_examples(db) -> None:
|
|
"""Examples of aggregate operations."""
|
|
# Get the collection
|
|
products_collection = db.get_collection("products")
|
|
|
|
# Calculate average price by category
|
|
pipeline = [
|
|
{
|
|
"$group": {
|
|
"_id": "$category",
|
|
"avg_price": {"$avg": "$price"},
|
|
"total_products": {"$sum": 1},
|
|
}
|
|
},
|
|
{"$sort": {"avg_price": -1}},
|
|
]
|
|
results = products_collection.aggregate(pipeline)
|
|
print("Category statistics:", list(results))
|
|
|
|
|
|
@MongoDBHandler.with_mongodb
|
|
def complex_query_example(db) -> None:
|
|
"""Example of a more complex query combining multiple operations."""
|
|
# Get the collection
|
|
users_collection = db.get_collection("users")
|
|
|
|
# Find active users who made purchases in last 30 days
|
|
pipeline = [
|
|
{
|
|
"$match": {
|
|
"status": "active",
|
|
"last_purchase": {
|
|
"$gte": arrow.now().shift(days=-30).datetime,
|
|
},
|
|
}
|
|
},
|
|
{
|
|
"$lookup": {
|
|
"from": "orders",
|
|
"localField": "_id",
|
|
"foreignField": "user_id",
|
|
"as": "recent_orders",
|
|
}
|
|
},
|
|
{
|
|
"$project": {
|
|
"username": 1,
|
|
"email": 1,
|
|
"total_orders": {"$size": "$recent_orders"},
|
|
"total_spent": {"$sum": "$recent_orders.amount"},
|
|
}
|
|
},
|
|
{"$sort": {"total_spent": -1}},
|
|
]
|
|
results = users_collection.aggregate(pipeline)
|
|
print("Active users with recent purchases:", list(results))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Example usage of all operations
|
|
insert_examples()
|
|
find_examples()
|
|
update_examples()
|
|
delete_examples()
|
|
aggregate_examples()
|
|
complex_query_example()
|