feat(connectors.docker): add docker_user connector data option

src/pyinfra/connectors/docker.py

@@ -29,12 +29,14 @@ class ConnectorData(TypedDict):
     docker_identifier: str
     docker_platform: str
     docker_architecture: str
+    docker_user: str
 
 
 connector_data_meta: dict[str, DataMeta] = {
     "docker_identifier": DataMeta("ID of container or image to start from"),
     "docker_platform": DataMeta("Platform to use for Docker image (e.g., linux/amd64)"),
     "docker_architecture": DataMeta("Architecture to use for Docker image (e.g., amd64, arm64)"),
+    "docker_user": DataMeta("User to execute commands as inside the container"),
 }
 
 
@@ -207,9 +209,15 @@ def run_shell_command(
         command = StringCommand(QuoteString(command))
 
         docker_flags = "-it" if local_arguments.get("_get_pty") else "-i"
+
+        docker_command_bits: list[str | QuoteString] = [self.docker_cmd, "exec"]
+
+        docker_user = self.data.get("docker_user")
+        if docker_user:
+            docker_command_bits.extend(["--user", QuoteString(docker_user)])
+
         docker_command = StringCommand(
-            self.docker_cmd,
-            "exec",
+            *docker_command_bits,
             docker_flags,
             container_id,
             "sh",

tests/test_connectors/test_docker.py

@@ -106,6 +106,34 @@ def test_run_shell_command(self):
             stdin=PIPE,
         )
 
+    def test_run_shell_command_with_user(self):
+        inventory = make_inventory(
+            hosts=(f"@{self.connector_name}/not-an-image",),
+            override_data={"docker_user": "myuser"},
+        )
+        State(inventory, Config())
+
+        command = "echo hi"
+        self.fake_popen_mock().returncode = 0
+
+        host = inventory.get_host(f"@{self.connector_name}/not-an-image")
+        host.connect()
+        out = host.run_shell_command(command, _get_pty=True)
+        assert out[0] is True
+
+        command = make_unix_command(command).get_raw_value()
+        command = shlex.quote(command)
+        docker_command = f"{self.cli_cmd} exec --user myuser -it containerid sh -c {command}"
+        shell_command = make_unix_command(docker_command).get_raw_value()
+
+        self.fake_popen_mock.assert_called_with(
+            shell_command,
+            shell=True,
+            stdout=PIPE,
+            stderr=PIPE,
+            stdin=PIPE,
+        )
+
     def test_run_shell_command_success_exit_codes(self):
         inventory = make_inventory(hosts=(f"@{self.connector_name}/not-an-image",))
         State(inventory, Config())