selfprivacy-rest-api/selfprivacy_api/migrations/__init__.py

56 lines
2 KiB
Python
Raw Normal View History

2022-02-16 13:03:38 +00:00
"""Migrations module.
Migrations module is introduced in v1.1.1 and provides one-shot
migrations which cannot be performed from the NixOS configuration file changes.
These migrations are checked and ran before every start of the API.
You can disable certain migrations if needed by creating an array
at api.skippedMigrations in userdata.json and populating it
with IDs of the migrations to skip.
Adding DISABLE_ALL to that array disables the migrations module entirely.
"""
from selfprivacy_api.migrations.check_for_failed_binds_migration import (
CheckForFailedBindsMigration,
)
from selfprivacy_api.utils import ReadUserData
from selfprivacy_api.migrations.fix_nixos_config_branch import FixNixosConfigBranch
2022-01-14 07:24:33 +00:00
from selfprivacy_api.migrations.create_tokens_json import CreateTokensJson
from selfprivacy_api.migrations.migrate_to_selfprivacy_channel import (
MigrateToSelfprivacyChannel,
)
2022-07-26 12:33:44 +00:00
from selfprivacy_api.migrations.mount_volume import MountVolume
migrations = [
FixNixosConfigBranch(),
CreateTokensJson(),
MigrateToSelfprivacyChannel(),
MountVolume(),
2022-09-22 16:41:48 +00:00
CheckForFailedBindsMigration(),
]
2022-01-11 05:41:25 +00:00
def run_migrations():
"""
Go over all migrations. If they are not skipped in userdata file, run them
if the migration needed.
"""
with ReadUserData() as data:
if "api" not in data:
skipped_migrations = []
elif "skippedMigrations" not in data["api"]:
skipped_migrations = []
else:
skipped_migrations = data["api"].get("skippedMigrations", [])
if "DISABLE_ALL" in skipped_migrations:
return
for migration in migrations:
if migration.get_migration_name() not in skipped_migrations:
try:
if migration.is_migration_needed():
2022-01-11 05:52:15 +00:00
migration.migrate()
2022-02-16 13:03:38 +00:00
except Exception as err:
print(f"Error while migrating {migration.get_migration_name()}")
2022-02-16 13:03:38 +00:00
print(err)
print("Skipping this migration")