def migrate()

in src/databao_context_engine/storage/migrate.py [0:0]

• 13 lines of code
• 6 McCabe index (conditional complexity)

def migrate(db_path: str | Path | None = None, migration_files: list[Path] | None = None) -> None:
    if migration_files is None:
        migration_files = [
            migration
            for migration in files("databao_context_engine.storage.migrations").iterdir()
            if isinstance(migration, Path) and ".sql" == migration.suffix
        ]

    db = Path(db_path or get_db_path()).expanduser().resolve()
    db.parent.mkdir(parents=True, exist_ok=True)
    logger.debug("Running migrations on database: %s", db)

    migration_manager = _MigrationManager(db, migration_files)
    migration_manager.migrate()
    logger.debug("Migration complete")