Implements common config migration functions

nucypher/config/migrations/__init__.py

@@ -2,17 +2,13 @@ from collections import OrderedDict
 
 from .configuration_v1_to_v2 import configuration_v1_to_v2
 from .configuration_v3_to_v4 import configuration_v3_to_v4
-from .configuration_v4_to_v5 import configuration_v4_to_v5
 from .configuration_v4_to_v6 import configuration_v4_to_v6
-from .configuration_v5_to_v6 import configuration_v5_to_v6
 
 MIGRATIONS = OrderedDict(
     {
         (1, 2): configuration_v1_to_v2,
         (2, 3): None,  # (no-op)
         (3, 4): configuration_v3_to_v4,
-        (4, 5): configuration_v4_to_v5,
         (4, 6): configuration_v4_to_v6,
-        (5, 6): configuration_v5_to_v6,
     }
 )

nucypher/config/migrations/_template.py

@@ -0,0 +1,16 @@
+from typing import Dict
+
+from nucypher.config.migrations.common import perform_migration
+
+
+def __migration(config: Dict) -> None:
+    pass  # apply migrations here
+
+
+def configuration_v1_to_v2(filepath) -> None:  # rename accordingly
+    perform_migration(
+        old_version=1,  # migrating from version
+        new_version=2,  # migrating to version
+        migration=__migration,
+        filepath=filepath,
+    )

nucypher/config/migrations/configuration_v1_to_v2.py

@@ -1,44 +1,18 @@
-import json
+from typing import Dict
-import os
 
-BACKUP_SUFFIX = ".old"
+from nucypher.config.migrations.common import perform_migration
-OLD_VERSION = 1
-NEW_VERSION = 2
 
 
-def configuration_v1_to_v2(filepath: str):
+def __migration(config: Dict) -> None:
-    # Read + deserialize
+    domains = config["domains"]
-    with open(filepath, "r") as file:
+    domain = domains[0]
-        contents = file.read()
+    if len(domains) > 1:
-    config = json.loads(contents)
+        print(f"Multiple domains configured, using the first one ({domain}).")
+    del config["domains"]
+    config["domain"] = domain
 
-    try:
-        # Check we have version 1 indeed
-        existing_version = config["version"]
-        if existing_version != OLD_VERSION:
-            raise RuntimeError(
-                f"Existing configuration is not version 1; Got version {existing_version}"
-            )
 
-        # Make a copy of the original file
+def configuration_v1_to_v2(filepath) -> None:
-        backup_filepath = str(filepath) + BACKUP_SUFFIX
+    perform_migration(
-        os.rename(filepath, backup_filepath)
+        old_version=1, new_version=2, migration=__migration, filepath=filepath
-        print(f"Backed up existing configuration to {backup_filepath}")
+    )
-
-        # Get current domain value
-        domains = config["domains"]
-        domain = domains[0]
-        if len(domains) > 1:
-            print(f"Multiple domains configured, using the first one ({domain}).")
-
-        # Apply updates
-        del config["domains"]  # deprecated
-        config["domain"] = domain
-        config["version"] = NEW_VERSION
-    except KeyError:
-        raise KeyError(f"Invalid {OLD_VERSION} configuration file.")
-
-    # Commit updates
-    with open(filepath, "w") as file:
-        file.write(json.dumps(config, indent=4))
-    print("OK! Migrated configuration file from v1 -> v2.")

nucypher/config/migrations/configuration_v3_to_v4.py

@@ -1,40 +1,15 @@
-import json
+from typing import Dict
-import os
 
-BACKUP_SUFFIX = ".old"
+from nucypher.config.migrations.common import perform_migration
-OLD_VERSION = 3
-NEW_VERSION = 4
 
 
-def configuration_v3_to_v4(filepath: str):
+def __migration(config: Dict) -> None:
-    # Read + deserialize
+    worker_address = config["worker_address"]
-    with open(filepath, "r") as file:
+    del config["worker_address"]  # deprecated
-        contents = file.read()
+    config["operator_address"] = worker_address
-    config = json.loads(contents)
 
-    try:
-        # Check we have version 1 indeed
-        existing_version = config["version"]
-        if existing_version != OLD_VERSION:
-            raise RuntimeError(
-                f"Existing configuration is not version {OLD_VERSION}; Got version {existing_version}"
-            )
 
-        # Make a copy of the original file
+def configuration_v3_to_v4(filepath) -> None:
-        backup_filepath = str(filepath) + BACKUP_SUFFIX
+    perform_migration(
-        os.rename(filepath, backup_filepath)
+        old_version=3, new_version=4, migration=__migration, filepath=filepath
-        print(f"Backed up existing configuration to {backup_filepath}")
+    )
-
-        # Apply updates
-        worker_address = config["worker_address"]
-        del config["worker_address"]  # deprecated
-        config["operator_address"] = worker_address
-        config["version"] = NEW_VERSION
-
-    except KeyError:
-        raise RuntimeError(f"Invalid {OLD_VERSION} configuration file.")
-
-    # Commit updates
-    with open(filepath, "w") as file:
-        file.write(json.dumps(config, indent=4))
-    print(f"OK! Migrated configuration file from v{OLD_VERSION} -> v{NEW_VERSION}.")

nucypher/config/migrations/configuration_v4_to_v5.py

@@ -1,38 +1,14 @@
-import json
+from typing import Dict
-import os
 
-BACKUP_SUFFIX = ".old"
+from nucypher.config.migrations.common import perform_migration
-OLD_VERSION = 4
-NEW_VERSION = 5
 
 
-def configuration_v4_to_v5(filepath: str):
+def __migration(config: Dict) -> None:
-    # Read + deserialize
+    del config["federated_only"]  # deprecated
-    with open(filepath, "r") as file:
+    del config["checksum_address"]
-        contents = file.read()
-    config = json.loads(contents)
 
-    try:
-        existing_version = config["version"]
-        if existing_version != OLD_VERSION:
-            raise RuntimeError(
-                f"Existing configuration is not version {OLD_VERSION}; Got version {existing_version}"
-            )
 
-        # Make a copy of the original file
+def configuration_v4_to_v5(filepath) -> None:
-        backup_filepath = str(filepath) + BACKUP_SUFFIX
+    perform_migration(
-        os.rename(filepath, backup_filepath)
+        old_version=4, new_version=5, migration=__migration, filepath=filepath
-        print(f"Backed up existing configuration to {backup_filepath}")
+    )
-
-        # Apply updates
-        del config["federated_only"]  # deprecated
-        del config["checksum_address"]
-        config["version"] = NEW_VERSION
-
-    except KeyError:
-        raise RuntimeError(f"Invalid {OLD_VERSION} configuration file.")
-
-    # Commit updates
-    with open(filepath, "w") as file:
-        file.write(json.dumps(config, indent=4))
-    print(f"OK! Migrated configuration file from v{OLD_VERSION} -> v{NEW_VERSION}.")

nucypher/config/migrations/configuration_v4_to_v6.py

@@ -1,52 +1,25 @@
-import json
+from typing import Dict
-import os
 
 from nucypher.blockchain.eth.networks import NetworksInventory
-
+from nucypher.config.migrations.common import perform_migration
-BACKUP_SUFFIX = ".old"
-OLD_VERSION = 4
-NEW_VERSION = 6
 
 
-def configuration_v4_to_v6(filepath: str):
+def __migration(config: Dict) -> None:
-    # Read + deserialize
+    del config["federated_only"]  # deprecated
-    with open(filepath, "r") as file:
+    del config["checksum_address"]
-        contents = file.read()
-    config = json.loads(contents)
 
-    try:
+    # Multichain support
-        existing_version = config["version"]
+    eth_provider = config["eth_provider_uri"]
-        if existing_version != OLD_VERSION:
+    eth_chain_id = NetworksInventory.get_ethereum_chain_id(config["domain"])
-            raise RuntimeError(
+    polygon_provider = config["payment_provider"]
-                f"Existing configuration is not version {OLD_VERSION}; Got version {existing_version}"
+    polygon_chain_id = NetworksInventory.get_polygon_chain_id(config["payment_network"])
-            )
+    config["condition_providers"] = {
+        eth_chain_id: [eth_provider],
+        polygon_chain_id: [polygon_provider],
+    }
 
-        # Make a copy of the original file
-        backup_filepath = str(filepath) + BACKUP_SUFFIX
-        os.rename(filepath, backup_filepath)
-        print(f"Backed up existing configuration to {backup_filepath}")
 
-        # Apply updates
+def configuration_v4_to_v6(filepath) -> None:
-        del config["federated_only"]  # deprecated
+    perform_migration(
-        del config["checksum_address"]
+        old_version=4, new_version=6, migration=__migration, filepath=filepath
-        config["version"] = NEW_VERSION
+    )
-
-        # Multichain support
-        eth_provider = config["eth_provider_uri"]
-        eth_chain_id = NetworksInventory.get_ethereum_chain_id(config["domain"])
-        polygon_provider = config["payment_provider"]
-        polygon_chain_id = NetworksInventory.get_polygon_chain_id(
-            config["payment_network"]
-        )
-        config["condition_providers"] = {
-            eth_chain_id: [eth_provider],
-            polygon_chain_id: [polygon_provider],
-        }
-
-    except KeyError:
-        raise RuntimeError(f"Invalid {OLD_VERSION} configuration file.")
-
-    # Commit updates
-    with open(filepath, "w") as file:
-        file.write(json.dumps(config, indent=4))
-    print(f"OK! Migrated configuration file from v{OLD_VERSION} -> v{NEW_VERSION}.")

nucypher/config/migrations/configuration_v5_to_v6.py

@@ -1,50 +1,21 @@
-import json
+from typing import Dict
-import os
 
 from nucypher.blockchain.eth.networks import NetworksInventory
-
+from nucypher.config.migrations.common import perform_migration
-BACKUP_SUFFIX = ".old"
-OLD_VERSION = 5
-NEW_VERSION = 6
 
 
-def configuration_v5_to_v6(filepath: str):
+def __migration(config: Dict) -> None:
-    # Read + deserialize
+    eth_provider = config["eth_provider_uri"]
-    with open(filepath, "r") as file:
+    eth_chain_id = NetworksInventory.get_ethereum_chain_id(config["domain"])
-        contents = file.read()
+    polygon_provider = config["payment_provider"]
-    config = json.loads(contents)
+    polygon_chain_id = NetworksInventory.get_polygon_chain_id(config["payment_network"])
+    config["condition_providers"] = {
+        eth_chain_id: [eth_provider],
+        polygon_chain_id: [polygon_provider],
+    }
 
-    try:
-        existing_version = config["version"]
-        if existing_version != OLD_VERSION:
-            raise RuntimeError(
-                f"Existing configuration is not version {OLD_VERSION}; Got version {existing_version}"
-            )
 
-        # Make a copy of the original file
+def configuration_v5_to_v6(filepath) -> None:
-        backup_filepath = str(filepath) + BACKUP_SUFFIX
+    perform_migration(
-        os.rename(filepath, backup_filepath)
+        old_version=5, new_version=6, migration=__migration, filepath=filepath
-        print(f"Backed up existing configuration to {backup_filepath}")
+    )
-
-        # Apply updates
-        config["version"] = NEW_VERSION
-
-        # Multichain support
-        eth_provider = config["eth_provider_uri"]
-        eth_chain_id = NetworksInventory.get_ethereum_chain_id(config["domain"])
-        polygon_provider = config["payment_provider"]
-        polygon_chain_id = NetworksInventory.get_polygon_chain_id(
-            config["payment_network"]
-        )
-        config["condition_providers"] = {
-            eth_chain_id: [eth_provider],
-            polygon_chain_id: [polygon_provider],
-        }
-
-    except KeyError:
-        raise KeyError(f"Invalid {OLD_VERSION} configuration file.")
-
-    # Commit updates
-    with open(filepath, "w") as file:
-        file.write(json.dumps(config, indent=4))
-    print(f"OK! Migrated configuration file from v{OLD_VERSION} -> v{NEW_VERSION}.")