Magic init for Beanie. I was inspired by zero-conf initialization in Redis OM.

magic_init.py

import inspect
import os
from importlib.util import spec_from_file_location, module_from_spec
from pathlib import Path

from beanie import Document, init_beanie
from motor.motor_asyncio import AsyncIOMotorClient


class MongoInitializer:
    async def run(self, database='hat'):
        models = list(self.__discover_models(Path('.')))
        conn_url = os.getenv('MONGO_BEANIE_URL', 'mongodb://localhost:27017')
        engine = AsyncIOMotorClient(conn_url)
        await init_beanie(database=engine[database], document_models=models)

    def __discover_models(self, path: Path, model_modules='models.py'):
        for sub in path.glob(f'**/{model_modules}'):  # find all models.py files
            model_pkg = sub.__str__().replace('\\', '.').replace('.py', '')  # convert hat/auth/models.py to hat.auth.models
            # load module
            spec = spec_from_file_location(model_pkg, sub)
            mod = module_from_spec(spec)
            spec.loader.exec_module(mod)

            # load only classes
            for name, cls in inspect.getmembers(mod, inspect.isclass):
                if issubclass(cls.__base__, Document):
                    yield f'{model_pkg}.{name}'