Tutorial: Building a MongoDB-Primary Framework-M Application

This tutorial demonstrates how to build a MongoDB-based application using Framework-M's MX (Modular eXtension) pattern. You'll learn how to use framework-mx-mongo to replace the default SQLAlchemy backend with MongoDB, without forking the framework.

What is the MX Pattern?

The MX pattern enables enterprises to customize Framework-M by:

• Pure package composition - Install packages to change behavior
• No forking required - Maintain upgrade path to official releases
• Entry point discovery - Framework auto-discovers adapters at runtime
• Protocol-based - All adapters implement standard interfaces

Prerequisites

• Python 3.12+
• MongoDB 4.4+ (local or remote instance)
• Basic understanding of Framework-M concepts

Step 1: Install Framework-M with MongoDB Support

Instead of the standard framework-m package, install the core package with MongoDB adapters:

# Option 1: Core + MongoDB (no SQLAlchemy)
pip install framework-m-core framework-mx-mongo

# Option 2: With development tools
pip install framework-m-core framework-mx-mongo framework-m-studio

What happens:

• framework-m-core provides protocols and interfaces
• framework-mx-mongo provides MongoDB implementations
• Entry points automatically override defaults

Step 2: Verify MongoDB Connection

Create a simple script to verify the setup:

# verify_mongo.py
import asyncio
from framework_mx_mongo import MongoSessionFactory

async def main():
    factory = MongoSessionFactory(
        connection_url="mongodb://localhost:27017",
        database_name="myapp"
    )

    # Test connection
    if await factory.ping():
        print("✓ MongoDB connected successfully!")
    else:
        print("✗ MongoDB connection failed")

    await factory.close()

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

Run it:

python verify_mongo.py

Step 3: Create a DocType

DocTypes work the same with MongoDB as with SQLAlchemy:

# models.py
from datetime import datetime
from framework_m_core.base_doctype import BaseDocTypeProtocol
from pydantic import BaseModel, Field
from uuid import uuid4

class User(BaseModel):
    """User DocType - stored in MongoDB."""

    id: str = Field(default_factory=lambda: str(uuid4()))
    name: str
    email: str
    age: int | None = None
    created_at: datetime = Field(default_factory=datetime.now)
    modified_at: datetime = Field(default_factory=datetime.now)

    class Config:
        # MongoDB-specific: collection name
        collection_name = "users"

Step 4: Use the Repository

The repository API is identical - protocol-based design ensures compatibility:

# app.py
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from framework_mx_mongo import MongoGenericRepository
from models import User

async def main():
    # Connect to MongoDB
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    db = client["myapp"]

    # Create repository
    user_repo = MongoGenericRepository(User, db)

    # Create a user
    new_user = User(name="Alice", email="alice@example.com", age=30)
    saved_user = await user_repo.save(new_user)
    print(f"Created user: {saved_user.id}")

    # Retrieve user
    from uuid import UUID
    found_user = await user_repo.get(UUID(saved_user.id))
    print(f"Found user: {found_user.name}")

    # List users with filtering
    from framework_m_core.interfaces.repository import FilterSpec, FilterOperator

    filters = [
        FilterSpec(field="age", operator=FilterOperator.GTE, value=25)
    ]
    result = await user_repo.list(filters=filters, limit=10)
    print(f"Found {result.total} users aged 25+")

    # Cleanup
    client.close()

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

Step 5: Configure Bootstrap (Optional)

For advanced applications, configure the bootstrap sequence:

# config.py
from framework_m_core.bootstrap import BootstrapRunner
from framework_mx_mongo import MongoInit

async def initialize_app():
    """Initialize application with MongoDB."""

    # Create DI container (simplified)
    container = {}  # Use real DI container in production

    # Run bootstrap
    runner = BootstrapRunner()
    await runner.run(container)

    # Bootstrap discovers and runs:
    # 1. MongoInit (order=10) - from framework-mx-mongo
    # 2. Any other registered steps

    print("Application initialized!")

# In your main app
import asyncio
asyncio.run(initialize_app())

Step 6: Schema Management

MongoDB schemas are defined via JSON Schema validators:

# schema.py
from framework_mx_mongo import MongoSchemaMapper
from models import User

async def setup_schema():
    """Create MongoDB collection with validation."""
    from motor.motor_asyncio import AsyncIOMotorClient

    client = AsyncIOMotorClient("mongodb://localhost:27017")
    db = client["myapp"]

    # Generate schema
    mapper = MongoSchemaMapper()
    schema = mapper.create_table(User, None)

    # Create collection with validation
    await db.create_collection(
        schema["collection"],
        validator=schema["validator"]
    )

    # Create indexes
    collection = db[schema["collection"]]
    for index_spec in schema["indexes"]:
        await collection.create_index(
            index_spec["keys"],
            unique=index_spec.get("unique", False)
        )

    print(f"Collection '{schema['collection']}' created with schema validation")
    client.close()

Step 7: Transactions (MongoDB 4.0+)

Use MongoDB sessions for ACID transactions:

# transactions.py
import asyncio
from framework_mx_mongo import MongoSessionFactory
from motor.motor_asyncio import AsyncIOMotorClient

async def transfer_credits():
    """Example: Transfer credits between users atomically."""

    factory = MongoSessionFactory(
        "mongodb://localhost:27017",
        "myapp"
    )

    async with factory.session() as session:
        # Operations within this block use the same session
        db = factory.get_database()
        users = db["users"]

        # Deduct from user A
        await users.update_one(
            {"id": "user-a-id"},
            {"$inc": {"credits": -100}},
            session=session
        )

        # Add to user B
        await users.update_one(
            {"id": "user-b-id"},
            {"$inc": {"credits": 100}},
            session=session
        )

        # Auto-commits on success, rolls back on exception

    await factory.close()
    print("Transaction completed!")

Step 8: Testing Your Application

Tests work the same - mock the MongoDB database:

# test_users.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from framework_mx_mongo import MongoGenericRepository
from models import User

@pytest.fixture
def mock_db():
    """Mock MongoDB database."""
    mock = MagicMock()
    mock_collection = MagicMock()
    mock_collection.find_one = AsyncMock(return_value=None)
    mock_collection.insert_one = AsyncMock()
    mock.__getitem__ = lambda self, name: mock_collection
    return mock

@pytest.mark.asyncio
async def test_create_user(mock_db):
    """Test user creation with MongoDB repository."""
    repo = MongoGenericRepository(User, mock_db)

    user = User(name="Test User", email="test@example.com")
    saved_user = await repo.save(user)

    assert saved_user.name == "Test User"
    mock_db["user"].insert_one.assert_called_once()

Migration from SQLAlchemy

If migrating an existing app from SQLAlchemy:

1. Update Dependencies

# Before
dependencies = ["framework-m"]

# After
dependencies = ["framework-m-core", "framework-mx-mongo"]

2. Update Imports

# Before
from framework_m.adapters.db import GenericRepository

# After - imports are the same, but implementation changes!
from framework_m_core.interfaces.repository import RepositoryProtocol
# Repository is auto-discovered from framework-mx-mongo

3. Data Migration

Export from PostgreSQL/MySQL and import to MongoDB:

# migrate_data.py
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from motor.motor_asyncio import AsyncIOMotorClient

async def migrate():
    # Read from SQLAlchemy
    sql_engine = create_async_engine("postgresql+asyncpg://...")

    # Write to MongoDB
    mongo_client = AsyncIOMotorClient("mongodb://localhost:27017")
    mongo_db = mongo_client["myapp"]

    # Migrate users table -> users collection
    async with sql_engine.connect() as conn:
        result = await conn.execute("SELECT * FROM users")
        users = result.fetchall()

        for user in users:
            await mongo_db.users.insert_one({
                "id": str(user.id),
                "name": user.name,
                "email": user.email,
                # ... other fields
            })

    print("Migration completed!")

asyncio.run(migrate())

Best Practices

1. Use Indexes Wisely

from pydantic import Field

class User(BaseModel):
    email: str = Field(
        ...,
        json_schema_extra={"index": True, "unique": True}
    )

2. Handle Connection Pooling

# Singleton pattern for MongoDB client
from motor.motor_asyncio import AsyncIOMotorClient

class MongoConnection:
    _client = None

    @classmethod
    def get_client(cls):
        if cls._client is None:
            cls._client = AsyncIOMotorClient(
                "mongodb://localhost:27017",
                maxPoolSize=50
            )
        return cls._client

3. Use Pydantic Validation

MongoDB stores JSON documents - leverage Pydantic for validation:

from pydantic import BaseModel, field_validator

class User(BaseModel):
    email: str

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email')
        return v.lower()

Comparison: Standard vs MX-Mongo

Feature	framework-m (Standard)	framework-mx-mongo
Database	PostgreSQL/MySQL (SQLAlchemy)	MongoDB (Motor)
Transactions	SQL transactions	MongoDB sessions
Schema	Tables + columns	Collections + documents
Migrations	Alembic	MongoDB schema versioning
Joins	SQL joins	Embedded docs / $lookup
Install size	Larger (SQLAlchemy)	Smaller (Motor only)

Advanced: Creating Your Own MX Package

Want to support Cassandra, DynamoDB, or another database? Follow the pattern:

framework-mx-cassandra/
├── pyproject.toml          # Register entry points
├── src/framework_mx_cassandra/
│   ├── repository.py       # Implement RepositoryProtocol
│   ├── schema_mapper.py    # Implement SchemaMapperProtocol
│   └── bootstrap.py        # Implement BootstrapProtocol

See the MX Pattern ADR for architectural details.

Troubleshooting

Connection Issues

# Test MongoDB connection
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017", serverSelectionTimeoutMS=5000)
client.admin.command('ping')  # Should not raise exception

Entry Point Not Found

Verify installation:

python -c "from importlib.metadata import entry_points; print(list(entry_points(group='framework_m.adapters.repository')))"

Schema Validation Errors

Check validator syntax:

schema = mapper.create_table(User, None)
print(schema["validator"])  # Inspect generated JSON Schema

Next Steps

• Read MX Pattern ADR
• Explore Protocol Reference
• Build your own MX package for your database!

Summary

You've learned how to:

• ✅ Install Framework-M with MongoDB support
• ✅ Create DocTypes that work with MongoDB
• ✅ Use repositories with the same API
• ✅ Handle transactions with MongoDB sessions
• ✅ Migrate from SQLAlchemy to MongoDB
• ✅ Understand the MX pattern benefits

The MX pattern gives you database flexibility without vendor lock-in!