import importlib.util from pathlib import Path import sqlite3 from typing import Callable def load_migrations(path: Path) -> list[tuple[int, str, Callable[[sqlite3.Connection], None]]]: migrations: list[tuple[int, str, Callable[[sqlite3.Connection], None]]] = [] migration_files = sorted( [f for f in path.glob("*.py") if not f.stem.startswith("_")] ) for filepath in migration_files: filename = filepath.stem version_str = filename.split("_")[0] try: version = int(version_str) except ValueError: raise ValueError('migrations must start with a number!!') spec = importlib.util.spec_from_file_location(filepath.stem, filepath) if not spec or not spec.loader: raise Exception(f"Failed to load spec from file: {filepath}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) if hasattr(module, "migrate"): migrations.append((version, filename, module.migrate)) else: raise ValueError(f"Migration {filepath.name} missing 'migrate' function") migrations.sort(key=lambda x: x[0]) return migrations