[Migrations] Filter to select migration scripts (#255)

Added a filter to run only specific migration scripts using a
glob expression. This is especially useful for development
purposes when testing a new script.

Author: odesenfans

deployment/migrations/config_updater.py

@@ -15,9 +15,10 @@
 import asyncio
 import importlib.util
 import logging
-import os
 import sys
+from pathlib import Path
 from types import ModuleType
+from typing import Iterable, Optional
 
 from configmanager import Config
 
@@ -61,6 +62,14 @@ def cli_parse() -> argparse.Namespace:
         type=str,
         help="Path to the private key file, if any. Only used to upgrade the key to the latest format.",
     )
+    parser.add_argument(
+        "--filter-scripts",
+        action="store",
+        required=False,
+        type=str,
+        help="A filter for migration scripts. If specified, only the files "
+        "matching the provided glob expression will be run.",
+    )
     parser.add_argument(
         "--verbose",
         "-v",
@@ -94,6 +103,18 @@ def import_module_from_path(path: str) -> ModuleType:
     return migration_module
 
 
+def list_migration_scripts(
+    migrations_dir: Path, glob_expression: Optional[str]
+) -> Iterable[Path]:
+    migration_scripts = set(migrations_dir.glob("*.py"))
+    if glob_expression:
+        migration_scripts = migration_scripts & set(
+            migrations_dir.glob(glob_expression)
+        )
+
+    return migration_scripts
+
+
 async def main(args: argparse.Namespace):
     log_level = logging.DEBUG if args.verbose else logging.INFO
     setup_logging(log_level)
@@ -102,16 +123,16 @@ async def main(args: argparse.Namespace):
     config = init_config(args.config)
     init_db_globals(config=config)
 
-    migration_scripts_dir = os.path.join(os.path.dirname(__file__), "scripts")
+    migration_scripts_dir = Path(__file__).parent / "scripts"
     migration_scripts = sorted(
-        f for f in os.listdir(migration_scripts_dir) if f.endswith(".py")
+        list_migration_scripts(migration_scripts_dir, args.filter_scripts)
     )
 
     command = args.command
 
     for migration_script in migration_scripts:
-        migration_script_path = os.path.join(migration_scripts_dir, migration_script)
-        migration_module = import_module_from_path(migration_script_path)
+        migration_script_path = migration_scripts_dir / migration_script
+        migration_module = import_module_from_path(str(migration_script_path))
 
         if args.verbose:
             LOGGER.info(f"%s: %s", migration_script, migration_module.__doc__)