Use Pydantic for validation and serialisation

.github/workflows/lint_code.yaml

@@ -59,6 +59,8 @@ jobs:
           python-version: 3.11
       - name: Install hatch
         run: pip install hatch
+      - name: Print Ruff version
+        run: hatch run lint:ruff --version
       - name: Lint Python
         run: hatch run lint:all
 

data_safe_haven/README.md

@@ -11,9 +11,18 @@ Install the following requirements before starting
 
 ```console
 > dsh context add ...
+> dsh context switch ...
 > dsh context create
 ```
 
+- Create the configuration
+
+```console
+> dsh config template --file config.yaml
+> vim config.yaml
+> dsh config upload config.yaml
+```
+
 - Next deploy the Safe Haven Management (SHM) infrastructure [approx 30 minutes]:
 
 ```console

data_safe_haven/administration/users/active_directory_users.py

@@ -24,7 +24,7 @@ def __init__(
     ) -> None:
         super().__init__(*args, **kwargs)
         shm_stack = SHMStackManager(config)
-        self.azure_api = AzureApi(config.subscription_name)
+        self.azure_api = AzureApi(config.context.subscription_name)
         self.logger = LoggingSingleton()
         self.resource_group_name = shm_stack.output("domain_controllers")[
             "resource_group_name"
@@ -77,7 +77,7 @@ def add(self, new_users: Sequence[ResearchUser]) -> None:
         for line in output.split("\n"):
             self.logger.parse(line)
 
-    def list(self, sre_name: str | None = None) -> Sequence[ResearchUser]:  # noqa: A003
+    def list(self, sre_name: str | None = None) -> Sequence[ResearchUser]:
         """List users in a local Active Directory"""
         list_users_script = FileReader(
             self.resources_path / "active_directory" / "list_users.ps1"
@@ -142,7 +142,7 @@ def remove(self, users: Sequence[ResearchUser]) -> None:
         for line in output.split("\n"):
             self.logger.parse(line)
 
-    def set(self, users: Sequence[ResearchUser]) -> None:  # noqa: A003
+    def set(self, users: Sequence[ResearchUser]) -> None:
         """Set local Active Directory users to specified list"""
         users_to_remove = [user for user in self.list() if user not in users]
         self.remove(users_to_remove)

data_safe_haven/administration/users/azure_ad_users.py

@@ -56,7 +56,7 @@ def add(self, new_users: Sequence[ResearchUser]) -> None:
         #     # Also add the user to the research users group
         #     self.graph_api.add_user_to_group(user.username, self.researchers_group_name)
 
-    def list(self) -> Sequence[ResearchUser]:  # noqa: A003
+    def list(self) -> Sequence[ResearchUser]:
         user_list = self.graph_api.read_users()
         return [
             ResearchUser(
@@ -105,7 +105,7 @@ def remove(self, users: Sequence[ResearchUser]) -> None:
         #         )
         pass
 
-    def set(self, users: Sequence[ResearchUser]) -> None:  # noqa: A003
+    def set(self, users: Sequence[ResearchUser]) -> None:
         """Set Guacamole users to specified list"""
         users_to_remove = [user for user in self.list() if user not in users]
         self.remove(users_to_remove)

data_safe_haven/administration/users/guacamole_users.py

@@ -18,7 +18,7 @@ def __init__(self, config: Config, sre_name: str, *args: Any, **kwargs: Any):
             sre_stack.secret("password-user-database-admin"),
             sre_stack.output("remote_desktop")["connection_db_server_name"],
             sre_stack.output("remote_desktop")["resource_group_name"],
-            config.subscription_name,
+            config.context.subscription_name,
         )
         self.users_: Sequence[ResearchUser] | None = None
         self.postgres_script_path: pathlib.Path = (
@@ -30,7 +30,7 @@ def __init__(self, config: Config, sre_name: str, *args: Any, **kwargs: Any):
         self.sre_name = sre_name
         self.group_name = f"Data Safe Haven SRE {sre_name} Users"
 
-    def list(self) -> Sequence[ResearchUser]:  # noqa: A003
+    def list(self) -> Sequence[ResearchUser]:
         """List all Guacamole users"""
         if self.users_ is None:  # Allow for the possibility of an empty list of users
             postgres_output = self.postgres_provisioner.execute_scripts(

data_safe_haven/administration/users/user_handler.py

@@ -97,7 +97,7 @@ def get_usernames_guacamole(self, sre_name: str) -> list[str]:
             self.logger.error(f"Could not load users for SRE '{sre_name}'.")
             return []
 
-    def list(self) -> None:  # noqa: A003
+    def list(self) -> None:
         """List Active Directory, AzureAD and Guacamole users
 
         Raises:
@@ -157,7 +157,7 @@ def remove(self, user_names: Sequence[str]) -> None:
             msg = f"Could not remove users: {user_names}.\n{exc}"
             raise DataSafeHavenUserHandlingError(msg) from exc
 
-    def set(self, users_csv_path: str) -> None:  # noqa: A003
+    def set(self, users_csv_path: str) -> None:
         """Set AzureAD and Guacamole users
 
         Raises:

data_safe_haven/cli.py

@@ -7,6 +7,7 @@
 from data_safe_haven import __version__
 from data_safe_haven.commands import (
     admin_command_group,
+    config_command_group,
     context_command_group,
     deploy_command_group,
     teardown_command_group,
@@ -69,6 +70,11 @@ def main() -> None:
         name="admin",
         help="Perform administrative tasks for a Data Safe Haven deployment.",
     )
+    application.add_typer(
+        config_command_group,
+        name="config",
+        help="Manage Data Safe Haven configuration.",
+    )
     application.add_typer(
         context_command_group, name="context", help="Manage Data Safe Haven contexts."
     )

data_safe_haven/commands/__init__.py

@@ -1,11 +1,13 @@
 from .admin import admin_command_group
+from .config import config_command_group
 from .context import context_command_group
 from .deploy import deploy_command_group
 from .teardown import teardown_command_group
 
 __all__ = [
     "admin_command_group",
     "context_command_group",
+    "config_command_group",
     "deploy_command_group",
     "teardown_command_group",
 ]

data_safe_haven/commands/admin_add_users.py

@@ -2,19 +2,19 @@
 import pathlib
 
 from data_safe_haven.administration.users import UserHandler
-from data_safe_haven.config import Config
+from data_safe_haven.config import Config, ContextSettings
 from data_safe_haven.exceptions import DataSafeHavenError
 from data_safe_haven.external import GraphApi
 
 
 def admin_add_users(csv_path: pathlib.Path) -> None:
     """Add users to a deployed Data Safe Haven"""
-    shm_name = "UNKNOWN"
-    try:
-        # Load config file
-        config = Config()
-        shm_name = config.name
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
+
+    shm_name = context.shm_name
 
+    try:
         # Load GraphAPI as this may require user-interaction that is not
         # possible as part of a Pulumi declarative command
         graph_api = GraphApi(

data_safe_haven/commands/admin_list_users.py

@@ -1,18 +1,18 @@
 """List users from a deployed Data Safe Haven"""
 from data_safe_haven.administration.users import UserHandler
-from data_safe_haven.config import Config
+from data_safe_haven.config import Config, ContextSettings
 from data_safe_haven.exceptions import DataSafeHavenError
 from data_safe_haven.external import GraphApi
 
 
 def admin_list_users() -> None:
     """List users from a deployed Data Safe Haven"""
-    shm_name = "UNKNOWN"
-    try:
-        # Load config file
-        config = Config()
-        shm_name = config.name
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
+
+    shm_name = context.shm_name
 
+    try:
         # Load GraphAPI as this may require user-interaction that is not
         # possible as part of a Pulumi declarative command
         graph_api = GraphApi(

data_safe_haven/commands/admin_register_users.py

@@ -1,6 +1,6 @@
 """Register existing users with a deployed SRE"""
 from data_safe_haven.administration.users import UserHandler
-from data_safe_haven.config import Config
+from data_safe_haven.config import Config, ContextSettings
 from data_safe_haven.exceptions import DataSafeHavenError
 from data_safe_haven.external import GraphApi
 from data_safe_haven.functions import alphanumeric
@@ -12,16 +12,14 @@ def admin_register_users(
     sre: str,
 ) -> None:
     """Register existing users with a deployed SRE"""
-    shm_name = "UNKNOWN"
-    sre_name = "UNKNOWN"
-    try:
-        # Use a JSON-safe SRE name
-        sre_name = alphanumeric(sre).lower()
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
 
-        # Load config file
-        config = Config()
-        shm_name = config.name
+    shm_name = context.shm_name
+    # Use a JSON-safe SRE name
+    sre_name = alphanumeric(sre).lower()
 
+    try:
         # Check that SRE option has been provided
         if not sre_name:
             msg = "SRE name must be specified."

data_safe_haven/commands/admin_remove_users.py

@@ -1,6 +1,6 @@
 """Remove existing users from a deployed Data Safe Haven"""
 from data_safe_haven.administration.users import UserHandler
-from data_safe_haven.config import Config
+from data_safe_haven.config import Config, ContextSettings
 from data_safe_haven.exceptions import DataSafeHavenError
 from data_safe_haven.external import GraphApi
 
@@ -9,12 +9,12 @@ def admin_remove_users(
     usernames: list[str],
 ) -> None:
     """Remove existing users from a deployed Data Safe Haven"""
-    shm_name = "UNKNOWN"
-    try:
-        # Load config file
-        config = Config()
-        shm_name = config.name
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
+
+    shm_name = context.shm_name
 
+    try:
         # Load GraphAPI as this may require user-interaction that is not
         # possible as part of a Pulumi declarative command
         graph_api = GraphApi(

data_safe_haven/commands/admin_unregister_users.py

@@ -1,6 +1,6 @@
 """Unregister existing users from a deployed SRE"""
 from data_safe_haven.administration.users import UserHandler
-from data_safe_haven.config import Config
+from data_safe_haven.config import Config, ContextSettings
 from data_safe_haven.exceptions import DataSafeHavenError
 from data_safe_haven.external import GraphApi
 from data_safe_haven.functions import alphanumeric
@@ -12,16 +12,14 @@ def admin_unregister_users(
     sre: str,
 ) -> None:
     """Unregister existing users from a deployed SRE"""
-    shm_name = "UNKNOWN"
-    sre_name = "UNKNOWN"
-    try:
-        # Use a JSON-safe SRE name
-        sre_name = alphanumeric(sre).lower()
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
 
-        # Load config file
-        config = Config()
-        shm_name = config.name
+    shm_name = context.shm_name
+    # Use a JSON-safe SRE name
+    sre_name = alphanumeric(sre).lower()
 
+    try:
         # Check that SRE option has been provided
         if not sre_name:
             msg = "SRE name must be specified."

data_safe_haven/commands/config.py

@@ -0,0 +1,47 @@
+"""Command group and entrypoints for managing DSH configuration"""
+from pathlib import Path
+from typing import Annotated, Optional
+
+import typer
+from rich import print
+
+from data_safe_haven.config import Config, ContextSettings
+
+config_command_group = typer.Typer()
+
+
+@config_command_group.command()
+def template(
+    file: Annotated[
+        Optional[Path],  # noqa: UP007
+        typer.Option(help="File path to write configuration template to."),
+    ] = None
+) -> None:
+    """Write a template Data Safe Haven configuration."""
+    context = ContextSettings.from_file().assert_context()
+    config = Config.template(context)
+    if file:
+        with open(file, "w") as outfile:
+            outfile.write(config.to_yaml())
+    else:
+        print(config.to_yaml())
+
+
+@config_command_group.command()
+def upload(
+    file: Annotated[Path, typer.Argument(help="Path to configuration file")]
+) -> None:
+    """Upload a configuration to the Data Safe Haven context"""
+    context = ContextSettings.from_file().assert_context()
+    with open(file) as config_file:
+        config_yaml = config_file.read()
+    config = Config.from_yaml(context, config_yaml)
+    config.upload()
+
+
+@config_command_group.command()
+def show() -> None:
+    """Print the configuration for the selected Data Safe Haven context"""
+    context = ContextSettings.from_file().assert_context()
+    config = Config.from_remote(context)
+    print(config.to_yaml())