Add example usage from Python

[ThibaultLemaire]

The README shows an example usage from the CLI, I thought it'd be nice to have an example of Python code too.

To show that you are delivering on

keeping it usable [...] as [a] python dependency

[Cajuteq]

Hi, kind of an old issue, but we wanted to execute our migrations to the latest each time our backend was rebuild. We use Fastapi with Odmantic and Pymongo, it was kind of difficult as Odmantic does not easily expose the MongoClient. I though I is worth sharing.

Here is our code :

# initial_data.py
import logging
from pymongo_migrate.mongo_migrate import MongoMigrate
from app.core.config import settings # a .env reading file allowing to get settings.MONGO_DATABASE as a str
from app.db.session import MongoDatabase, _MongoClientSingleton

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

[...]

async def main() -> None:
    db = MongoDatabase() # initialize the mongoCLient using Odmantic AIOEngine
    logger.info("Creating initial data")

    await populate_db(db) # a coroutine doing some initial data creation if the db is empty (create admin user and some examples)
    logger.info("Initial data created")
    # the next two ligns are equivalent of CLI $ pymongo-migrate migrate -u 'mongodb://localhost:27017' -d 'app' -m 'pymongo_migrations' -c 'migrations_states'
    mongo_migrate = MongoMigrate(client=_MongoClientSingleton().mongo_client.delegate, database=settings.MONGO_DATABASE, migrations_dir="./pymongo_migrations/", migrations_collection="migrations_states", logger=logger)

    logger.info(mongo_migrate.migrate()) # apply the needed migrations

    logger.info("Data migration executed")

# app/db/session.py
from motor import core, motor_asyncio
from odmantic import AIOEngine
from pymongo.driver_info import DriverInfo

from app.__version__ import __version__
from app.core.config import settings

DRIVER_INFO = DriverInfo(name="full-stack-fastapi-mongodb", version=__version__)


class _MongoClientSingleton:
    mongo_client: motor_asyncio.AsyncIOMotorClient | None
    engine: AIOEngine

    def __new__(cls):
        if not hasattr(cls, "instance"):
            cls.instance = super(_MongoClientSingleton, cls).__new__(cls)
            cls.instance.mongo_client = motor_asyncio.AsyncIOMotorClient(
                settings.MONGO_DATABASE_URI, driver=DRIVER_INFO
            )
            cls.instance.engine = AIOEngine(client=cls.instance.mongo_client, database=settings.MONGO_DATABASE)
        return cls.instance


def MongoDatabase() -> core.AgnosticDatabase:
    return _MongoClientSingleton().mongo_client[settings.MONGO_DATABASE]

If you use MongoCLient directly (without Odmantic) and the default dir, collection and logger, the important ligns are as follow

from pymongo import MongoClient
from pymongo_migrate.mongo_migrate import MongoMigrate

c = MongoClient()
mongo_migrate = MongoMigrate(client=c, database='app')

mongo_migrate.migrate() # apply the needed migrations