def init_database()

in atr/db/__init__.py [0:0]
34 lines of code
3 McCabe index (conditional complexity)

def init_database(app: base.QuartApp) -> None:
    """
    Creates and initializes the database for a QuartApp.

    The database is created and an AsyncSession is registered as extension for the app.
    Any pending migrations are executed.
    """

    @app.before_serving
    async def create() -> None:
        global _global_atr_engine, _global_atr_sessionmaker

        app_config = config.get()
        engine = await create_async_engine(app_config)
        _global_atr_engine = engine

        _global_atr_sessionmaker = sqlalchemy.ext.asyncio.async_sessionmaker(
            bind=engine, class_=Session, expire_on_commit=False
        )

        # Run any pending migrations on startup
        _LOGGER.info("Applying database migrations via init_database...")
        alembic_ini_path = os.path.join(app_config.PROJECT_ROOT, "alembic.ini")
        alembic_cfg = Config(alembic_ini_path)

        # Construct synchronous URLs
        absolute_db_path = os.path.join(app_config.STATE_DIR, app_config.SQLITE_DB_PATH)
        sync_sqlalchemy_url = f"sqlite:///{absolute_db_path}"
        _LOGGER.info(f"Setting Alembic URL for command: {sync_sqlalchemy_url}")
        alembic_cfg.set_main_option("sqlalchemy.url", sync_sqlalchemy_url)

        # Ensure that Alembic finds the migrations directory relative to project root
        migrations_dir_path = os.path.join(app_config.PROJECT_ROOT, "migrations")
        _LOGGER.info(f"Setting Alembic script_location for command: {migrations_dir_path}")
        alembic_cfg.set_main_option("script_location", migrations_dir_path)

        try:
            _LOGGER.info("Running alembic upgrade head...")
            command.upgrade(alembic_cfg, "head")
            _LOGGER.info("Database migrations applied successfully")
        except Exception:
            _LOGGER.exception("Failed to apply database migrations during startup")
            raise

        try:
            _LOGGER.info("Running alembic check...")
            command.check(alembic_cfg)
            _LOGGER.info("Alembic check passed: DB schema matches models")
        except Exception:
            _LOGGER.exception("Failed to check database migrations during startup")
            raise