quaestor

In ancient Rome, a quaestor was the official in charge of the treasury — here, it manages your MongoDB schema and data, your real treasure.

Lightweight MongoDB migration module with named migrations and dependency resolution.

Install

pip install mongo-quaestor

Quick Start

import asyncio

from motor.motor_asyncio import AsyncIOMotorClient
from quaestor import MigrationRegistry

registry = MigrationRegistry()


@registry.register("create_users_index")
async def create_users_index(db):
    await db.users.create_index("email", unique=True)


@registry.register("add_status_field", depends_on=["create_users_index"])
async def add_status_field(db):
    await db.users.update_many(
        {"status": {"$exists": False}},
        {"$set": {"status": "active"}},
    )


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

    applied = await registry.run(db, "myapp")
    print(f"Applied migrations: {applied}")


asyncio.run(main())

API Reference

How It Works

• Applied migrations are tracked in a MongoDB collection named {prefix}_migrations.
• Each record stores name (str) and appliedAt (UTC datetime).
• Dependencies are resolved via topological sort (Kahn's algorithm). Circular or missing dependencies raise ValueError.
• Failed migrations are not recorded, so they retry automatically on the next run() call.
• Independent migrations (no dependency relationship between them) execute in alphabetical order for deterministic behavior.
• Multiple registries can safely share one database by using different prefixes.
• Progress and errors are logged to the quaestor logger.