mongoflow 0.1.0

Elegant MongoDB Object Document Mapper (ODM) for Python - fluent query builder included

MongoFlow 🌊

Elegant MongoDB Object Document Mapper (ODM) for Python - with a fluent query builder that makes working with MongoDB a breeze! 🚀

✨ Features

• 🎯 Intuitive Query Builder - Fluent, chainable queries that feel natural
• ⚡ High Performance - Connection pooling, batch operations, and streaming
• 🔧 Flexible - Use as simple queries or full ODM with models
• 🎨 Clean API - Pythonic, fully typed, and well-documented
• 🚀 Production Ready - Battle-tested patterns with automatic retries
• 💾 Smart Caching - Optional Redis integration for blazing speed
• 🔄 Async Support - Full async/await support with Motor
• 📦 Lightweight - Minimal dependencies, maximum functionality

📦 Installation

# Basic installation
pip install mongoflow

# With all features
pip install mongoflow[all]

# With specific features
pip install mongoflow[cache]      # Redis caching
pip install mongoflow[validation] # Pydantic validation
pip install mongoflow[async]      # Async support

🚀 Quick Start

Basic Usage

from mongoflow import MongoFlow, Repository

# Connect to MongoDB
MongoFlow.connect('mongodb://localhost:27017', 'mydb')

# Define a repository
class UserRepository(Repository):
    collection_name = 'users'

# Use it!
users = UserRepository()

# Create
user = users.create({
    'name': 'John Doe',
    'email': '[e-mail]',
    'age': 30
})

# Update or create
user, created = users.update_or_create(
    {'email': '[e-mail]'},  # Search criteria
    {
        'email': '[e-mail]',
        'name': 'Jane Doe',
        'age': 28,
        'status': 'active'
    }  # Full data to insert or update
)

# Query with fluent builder
active_adults = (users.query()
    .where('status', 'active')
    .where_greater_than('age', 18)
    .order_by('created_at', 'desc')
    .limit(10)
    .get())

Async Support

import asyncio
from mongoflow import AsyncMongoFlow, AsyncRepository

class AsyncUserRepository(AsyncRepository):
    collection_name = 'users'

async def main():
    # Connect
    await AsyncMongoFlow.connect('mongodb://localhost:27017', 'mydb')
    
    repo = AsyncUserRepository()
    
    # All methods support async/await
    user = await repo.create({'name': 'Jane'})
    
    # Update or create (async)
    user, created = await repo.update_or_create(
        {'code': 123},  # Search by code
        {
            'code': 123,
            'name': 'Async User',
            'email': '[e-mail]',
            'status': 'active'
        }  # Full data
    )
    
    query = await repo.query()  # Note: query() is async
    users = await query.where('active', True).get()
    
    # Async streaming for large datasets
    query = await repo.query()
    async for user in query.stream():
        print(user)

asyncio.run(main())

📚 Query Builder Methods

Both sync and async query builders support the following methods:

Filtering Methods

# Basic where conditions
.where('field', 'value')                    # field == value
.where('field', 'value', '$ne')            # field != value
.where_in('field', [1, 2, 3])              # field in [1, 2, 3]
.where_not_in('field', [1, 2, 3])          # field not in [1, 2, 3]
.where_between('field', min, max)          # min <= field <= max
.where_greater_than('field', value)        # field > value
.where_less_than('field', value)           # field < value
.where_like('field', 'pattern')            # regex pattern matching
.where_null('field')                        # field is null
.where_not_null('field')                    # field is not null
.where_exists('field', True)               # field exists in document

# Logical operators
.or_where([{'field1': 'value1'}, {'field2': 'value2'}])
.and_where([{'field1': 'value1'}, {'field2': 'value2'}])

Query Modifiers

# Projection
.select('field1', 'field2')                # Include only these fields
.exclude('field1', 'field2')               # Exclude these fields

# Sorting and pagination
.order_by('field', 'asc')                  # Sort ascending
.order_by('field', 'desc')                 # Sort descending
.skip(10)                                   # Skip N documents
.limit(20)                                  # Limit to N documents
.paginate(page=2, per_page=20)            # Get paginated results

Execution Methods

# Sync methods
.get()                                      # Get all results
.first()                                    # Get first result
.last()                                     # Get last result
.count()                                    # Count matching documents
.exists()                                   # Check if any documents exist
.distinct('field')                          # Get distinct values
.stream(batch_size=100)                    # Stream results

# Async methods (same as above but with await)
await query.get()
await query.first()
await query.last()
await query.count()
await query.exists()
await query.distinct('field')
async for doc in query.stream():
    process(doc)

Aggregation Methods

# Simple aggregations
.sum('field')                               # Sum of field values
.avg('field')                               # Average of field values
.min('field')                               # Minimum value
.max('field')                               # Maximum value

# Group operations
.group('status', {'count': {'$sum': 1}})   # Group by field
.group(
    {'status': '$status', 'type': '$type'},  # Group by multiple fields
    {'total': {'$sum': '$amount'}}           # With aggregations
)

# Custom aggregation pipelines
.aggregate([
    {'$match': {'status': 'active'}},
    {'$group': {'_id': '$category', 'total': {'$sum': '$amount'}}}
])

Async-Specific Features

All the above methods work with async query builders, just remember to use await:

# Async examples
query = await users.query()  # query() is async in AsyncRepository
active = await query.where('active', True).get()

query = await users.query()
count = await query.where_between('age', 18, 65).count()

query = await users.query()
stats = await query.group('role', {'count': {'$sum': 1}})

# Async streaming
query = await users.query()
async for user in query.where('status', 'active').stream():
    await process_user(user)