aboutsummaryrefslogtreecommitdiff
path: root/gn3/migrations.py
blob: 7f8d694d8897e45e13134ccb0623e1d539c355a2 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
"""Run the migrations in the app, rather than with yoyo CLI."""
from pathlib import Path
from typing import Union

from yoyo.backends import DatabaseBackend
from yoyo import read_migrations
from yoyo.migrations import Migration, MigrationList

class MigrationNotFound(Exception):
    """Raised if a migration is not found at the given path."""
    def __init__(self, migration_path: Path):
        """Initialise the exception."""
        super().__init__(f"Could not find migration '{migration_path}'")

def apply_migrations(backend: DatabaseBackend, migrations: MigrationList):
    "Apply the provided migrations."
    with backend.lock():
        backend.apply_migrations(backend.to_apply(migrations))

def rollback_migrations(backend: DatabaseBackend, migrations: MigrationList):
    "Rollback the provided migrations."
    with backend.lock():
        backend.rollback_migrations(backend.to_rollback(migrations))

def get_migration(migration_path: Union[Path, str]) -> Migration:
    """Retrieve a migration at thi given `migration_path`."""
    migration_path = Path(migration_path)
    if migration_path.exists():
        for migration in read_migrations(str(migration_path.parent)):
            if Path(migration.path) == migration_path:
                return migration

    raise MigrationNotFound(migration_path)