about summary refs log tree commit diff
path: root/gn_auth/migrations/__init__.py
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2026-05-21 13:32:56 -0500
committerFrederick Muriuki Muriithi2026-05-21 13:32:56 -0500
commit6568edd6bf616a0c17c5226f40b494a6b2967dad (patch)
treeccb986a3cec72061f490a15a154b299e12a13101 /gn_auth/migrations/__init__.py
parentdaf8d4ec95bb5d4a7e0c108c023f64d6954d70df (diff)
downloadgn-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__.py34
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)