aboutsummaryrefslogtreecommitdiff
path: root/gn3/migrations.py
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2022-11-03 09:05:44 +0300
committerFrederick Muriuki Muriithi2022-11-03 09:25:50 +0300
commitef70ddc1fb2800e340de50bcdb3cef2d34cc3b11 (patch)
tree1a3735ad4db6f567c9f70f8160009e16b82e9019 /gn3/migrations.py
parent6e58cb5ffe3b145aa6ba9b7a54672e7a7f18a0b7 (diff)
downloadgenenetwork3-ef70ddc1fb2800e340de50bcdb3cef2d34cc3b11.tar.gz
Initialise the Auth(entic|oris)ation packages
Initialise the authentication/authorisation system packages and set up the initial database migrations to set up the system. * README.md: Add documentation on migrations * gn3/auth/__init__.py: init package * gn3/auth/authentication/__init__.py: init package * gn3/auth/authorisation/__init__.py: init package * gn3/migrations.py: provide migration utilities * migrations/auth/20221103_01_js9ub-initialise-the-auth-entic-oris-ation-database.py: new migration * tests/unit/auth/test_init_database.py: test new migration applies and rolls back as expected * tests/unit/conftest.py: fixtures for unit tests * yoyo.auth.ini: basic configuration for yoyo-migration for auth system migrations
Diffstat (limited to 'gn3/migrations.py')
-rw-r--r--gn3/migrations.py33
1 files changed, 33 insertions, 0 deletions
diff --git a/gn3/migrations.py b/gn3/migrations.py
new file mode 100644
index 0000000..7f8d694
--- /dev/null
+++ b/gn3/migrations.py
@@ -0,0 +1,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)