Skip to content

mongrator.loader

mongrator.loader

load(config)

Scan migrations_dir, import each .py file, and return an ordered list.

Files are sorted lexicographically by filename, which is chronological when the recommended {timestamp}_{slug}.py naming convention is used.

Raises:

Type Description
DuplicateMigrationIdError

if two files share the same stem.

MigrationImportError

if a file cannot be imported.

InvalidMigrationFileError

if a file does not define an up() callable.

Source code in src/mongrator/loader.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def load(config: MigratorConfig) -> list[MigrationFile]:
    """Scan migrations_dir, import each .py file, and return an ordered list.

    Files are sorted lexicographically by filename, which is chronological
    when the recommended {timestamp}_{slug}.py naming convention is used.

    Raises:
        DuplicateMigrationIdError: if two files share the same stem.
        MigrationImportError: if a file cannot be imported.
        InvalidMigrationFileError: if a file does not define an up() callable.
    """
    migrations_dir = config.migrations_dir
    if not migrations_dir.exists():
        return []

    paths = sorted(migrations_dir.glob("*.py"))
    seen: dict[MigrationId, Path] = {}
    results: list[MigrationFile] = []

    for path in paths:
        migration_id = _migration_id(path)

        if migration_id in RESERVED_IDS:
            raise ReservedMigrationIdError(migration_id)
        if migration_id in seen:
            raise DuplicateMigrationIdError(migration_id)
        seen[migration_id] = path

        checksum = _checksum(path)
        module = _import_file(path, migration_id)

        if not callable(getattr(module, "up", None)):
            raise InvalidMigrationFileError(str(path), "missing a callable up() function")

        results.append(MigrationFile(id=migration_id, path=path, checksum=checksum, module=module))

    return results