Expand on the typing for volume types to support Kubernetes (#13467)

simonrw · web-flow · commit 9a9336f9edb9 · 2025-12-05T16:29:25.000Z
diff --git a/localstack-core/localstack/utils/bootstrap.py b/localstack-core/localstack/utils/bootstrap.py
@@ -34,8 +34,8 @@
     NoSuchImage,
     NoSuchNetwork,
     PortMappings,
-    VolumeDirMount,
     VolumeMappings,
+    VolumeMappingSpecification,
 )
 from localstack.utils.container_utils.docker_cmd_client import CmdDockerClient
 from localstack.utils.docker_utils import DOCKER_CLIENT
@@ -666,7 +666,7 @@ def _cfg(cfg: ContainerConfiguration):
         return _cfg
 
     @staticmethod
-    def volume(volume: BindMount | VolumeDirMount):
+    def volume(volume: VolumeMappingSpecification):
         def _cfg(cfg: ContainerConfiguration):
             cfg.volumes.add(volume)
 
diff --git a/localstack-core/localstack/utils/container_utils/container_client.py b/localstack-core/localstack/utils/container_utils/container_client.py
@@ -15,6 +15,7 @@
     Literal,
     NamedTuple,
     Protocol,
+    TypeAlias,
     TypedDict,
 )
 
@@ -357,7 +358,13 @@ def __repr__(self):
 
 
 @dataclasses.dataclass
-class BindMount:
+class Mount:
+    def to_str(self) -> str:
+        return str(self)
+
+
+@dataclasses.dataclass
+class BindMount(Mount):
     """Represents a --volume argument run/create command. When using VolumeBind to bind-mount a file or directory
     that does not yet exist on the Docker host, -v creates the endpoint for you. It is always created as a directory.
     """
@@ -402,7 +409,7 @@ def parse(cls, param: str) -> "BindMount":
 
 
 @dataclasses.dataclass
-class VolumeDirMount:
+class VolumeDirMount(Mount):
     volume_path: str
     """
     Absolute path inside /var/lib/localstack to mount into the container
@@ -439,24 +446,25 @@ def to_docker_sdk_parameters(self) -> tuple[str, dict[str, str]]:
         }
 
 
+VolumeMappingSpecification: TypeAlias = SimpleVolumeBind | Mount
+
+
 class VolumeMappings:
-    mappings: list[SimpleVolumeBind | BindMount]
+    mappings: list[VolumeMappingSpecification]
 
-    def __init__(self, mappings: list[SimpleVolumeBind | BindMount | VolumeDirMount] = None):
+    def __init__(
+        self,
+        mappings: list[VolumeMappingSpecification] = None,
+    ):
         self.mappings = mappings if mappings is not None else []
 
-    def add(self, mapping: SimpleVolumeBind | BindMount | VolumeDirMount):
+    def add(self, mapping: VolumeMappingSpecification):
         self.append(mapping)
 
-    def append(
-        self,
-        mapping: SimpleVolumeBind | BindMount | VolumeDirMount,
-    ):
+    def append(self, mapping: VolumeMappingSpecification):
         self.mappings.append(mapping)
 
-    def find_target_mapping(
-        self, container_dir: str
-    ) -> SimpleVolumeBind | BindMount | VolumeDirMount | None:
+    def find_target_mapping(self, container_dir: str) -> VolumeMappingSpecification | None:
         """
         Looks through the volumes and returns the one where the container dir matches ``container_dir``.
         Returns None if there is no volume mapping to the given container directory.
@@ -1510,7 +1518,8 @@ def convert_mount_list_to_dict(
     ) -> dict[str, dict[str, str]]:
         """Converts a List of (host_path, container_path) tuples to a Dict suitable as volume argument for docker sdk"""
 
-        def _map_to_dict(paths: SimpleVolumeBind | BindMount | VolumeDirMount):
+        def _map_to_dict(paths: VolumeMappingSpecification):
+            # TODO: move this logic to the `Mount` base class
             if isinstance(paths, (BindMount, VolumeDirMount)):
                 return paths.to_docker_sdk_parameters()
             else:
diff --git a/localstack-core/localstack/utils/container_utils/docker_cmd_client.py b/localstack-core/localstack/utils/container_utils/docker_cmd_client.py
@@ -12,7 +12,6 @@
 from localstack.utils.collections import ensure_list
 from localstack.utils.container_utils.container_client import (
     AccessDenied,
-    BindMount,
     CancellableStream,
     ContainerClient,
     ContainerException,
@@ -21,16 +20,16 @@
     DockerNotAvailable,
     DockerPlatform,
     LogConfig,
+    Mount,
     NoSuchContainer,
     NoSuchImage,
     NoSuchNetwork,
     NoSuchObject,
     PortMappings,
     RegistryConnectionError,
-    SimpleVolumeBind,
     Ulimit,
     Util,
-    VolumeDirMount,
+    VolumeMappingSpecification,
 )
 from localstack.utils.run import run
 from localstack.utils.strings import first_char_to_upper, to_str
@@ -810,7 +809,7 @@ def _build_run_create_cmd(
         tty: bool = False,
         detach: bool = False,
         command: list[str] | str | None = None,
-        volumes: list[SimpleVolumeBind] | None = None,
+        volumes: list[VolumeMappingSpecification] | None = None,
         ports: PortMappings | None = None,
         exposed_ports: list[str] | None = None,
         env_vars: dict[str, str] | None = None,
@@ -900,7 +899,7 @@ def _build_run_create_cmd(
         return cmd, env_file
 
     @staticmethod
-    def _map_to_volume_param(volume: SimpleVolumeBind | BindMount | VolumeDirMount) -> str:
+    def _map_to_volume_param(volume: VolumeMappingSpecification) -> str:
         """
         Maps the mount volume, to a parameter for the -v docker cli argument.
 
@@ -911,7 +910,8 @@ def _map_to_volume_param(volume: SimpleVolumeBind | BindMount | VolumeDirMount)
         :param volume: Either a SimpleVolumeBind, in essence a tuple (host_dir, container_dir), or a VolumeBind object
         :return: String which is passable as parameter to the docker cli -v option
         """
-        if isinstance(volume, (BindMount, VolumeDirMount)):
+        # TODO: move this logic to the VolumeMappingSpecification type
+        if isinstance(volume, Mount):
             return volume.to_str()
         else:
             return f"{volume[0]}:{volume[1]}"
