social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import importlib.util
2from pathlib import Path
3import sqlite3
4from typing import Callable
5
6
7def load_migrations(path: Path) -> list[tuple[int, str, Callable[[sqlite3.Connection], None]]]:
8 migrations: list[tuple[int, str, Callable[[sqlite3.Connection], None]]] = []
9 migration_files = sorted(
10 [f for f in path.glob("*.py") if not f.stem.startswith("_")]
11 )
12
13 for filepath in migration_files:
14 filename = filepath.stem
15 version_str = filename.split("_")[0]
16
17 try:
18 version = int(version_str)
19 except ValueError:
20 raise ValueError('migrations must start with a number!!')
21
22 spec = importlib.util.spec_from_file_location(filepath.stem, filepath)
23 if not spec or not spec.loader:
24 raise Exception(f"Failed to load spec from file: {filepath}")
25
26 module = importlib.util.module_from_spec(spec)
27 spec.loader.exec_module(module)
28
29 if hasattr(module, "migrate"):
30 migrations.append((version, filename, module.migrate))
31 else:
32 raise ValueError(f"Migration {filepath.name} missing 'migrate' function")
33
34 migrations.sort(key=lambda x: x[0])
35 return migrations