diff options
| author | Frederick Muriuki Muriithi | 2026-05-21 13:32:56 -0500 |
|---|---|---|
| committer | Frederick Muriuki Muriithi | 2026-05-21 13:32:56 -0500 |
| commit | 6568edd6bf616a0c17c5226f40b494a6b2967dad (patch) | |
| tree | ccb986a3cec72061f490a15a154b299e12a13101 /gn_auth/migrations/__init__.py | |
| parent | daf8d4ec95bb5d4a7e0c108c023f64d6954d70df (diff) | |
| download | gn-auth-6568edd6bf616a0c17c5226f40b494a6b2967dad.tar.gz | |
Move migrations to top-level gn_auth package.
In preparation for migrating to pyproject.toml (from setup.py and friends) we need to have only one top-level package. This will also help in improving testing and checks down the line, since everything will be relative to one single top-level directory.
Diffstat (limited to 'gn_auth/migrations/__init__.py')
| -rw-r--r-- | gn_auth/migrations/__init__.py | 34 |
1 files changed, 34 insertions, 0 deletions
diff --git a/gn_auth/migrations/__init__.py b/gn_auth/migrations/__init__.py new file mode 100644 index 0000000..6acb058 --- /dev/null +++ b/gn_auth/migrations/__init__.py @@ -0,0 +1,34 @@ +"""Migrations package: Provides the migrations, and some utility functions to +help with dealing with migrations.""" +from pathlib import Path +from typing import Union + +from yoyo import read_migrations +from yoyo.backends import DatabaseBackend +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) |
