feat: Add ContainerBackend with Docker and Podman

README.md

@@ -71,6 +71,29 @@ TrainerClient().wait_for_job_status(job_id)
 print("\n".join(TrainerClient().get_job_logs(name=job_id)))
 ```
 
+## Local Development
+
+Kubeflow Trainer client supports local development without needing a Kubernetes cluster.
+
+### Available Backends
+
+- **KubernetesBackend** (default) - Production training on Kubernetes
+- **ContainerBackend** - Local development with Docker/Podman isolation  
+- **LocalProcessBackend** - Quick prototyping with Python subprocesses
+
+**Quick Start:**
+Install container support: `pip install kubeflow[docker]` or `pip install kubeflow[podman]`
+
+```python
+from kubeflow.trainer import TrainerClient, ContainerBackendConfig, CustomTrainer
+
+# Switch to local container execution
+client = TrainerClient(backend_config=ContainerBackendConfig())
+
+# Your training runs locally in isolated containers
+job_id = client.train(trainer=CustomTrainer(func=train_fn))
+```
+
 ## Supported Kubeflow Projects
 
 | Project                     | Status           | Version Support | Description                                                          |

kubeflow/trainer/__init__.py

@@ -18,6 +18,10 @@
 
 # Import the Kubeflow Trainer client.
 from kubeflow.trainer.api.trainer_client import TrainerClient
+from kubeflow.trainer.backends.container.types import (
+    ContainerBackendConfig,
+    TrainingRuntimeSource,
+)
 from kubeflow.trainer.backends.localprocess.types import LocalProcessBackendConfig
 
 # Import the Kubeflow Trainer constants.
@@ -64,5 +68,7 @@
     "TrainerClient",
     "TrainerType",
     "LocalProcessBackendConfig",
+    "ContainerBackendConfig",
     "KubernetesBackendConfig",
+    "TrainingRuntimeSource",
 ]

kubeflow/trainer/api/trainer_client.py

@@ -17,6 +17,8 @@
 from typing import Optional, Union
 
 from kubeflow.common.types import KubernetesBackendConfig
+from kubeflow.trainer.backends.container.backend import ContainerBackend
+from kubeflow.trainer.backends.container.types import ContainerBackendConfig
 from kubeflow.trainer.backends.kubernetes.backend import KubernetesBackend
 from kubeflow.trainer.backends.localprocess.backend import (
     LocalProcessBackend,
@@ -31,14 +33,19 @@
 class TrainerClient:
     def __init__(
         self,
-        backend_config: Optional[Union[KubernetesBackendConfig, LocalProcessBackendConfig]] = None,
+        backend_config: Union[
+            KubernetesBackendConfig,
+            LocalProcessBackendConfig,
+            ContainerBackendConfig,
+        ] = None,
     ):
         """Initialize a Kubeflow Trainer client.
 
         Args:
-            backend_config: Backend configuration. Either KubernetesBackendConfig or
-                            LocalProcessBackendConfig, or None to use the backend's
-                            default config class. Defaults to KubernetesBackendConfig.
+            backend_config: Backend configuration. Either KubernetesBackendConfig,
+                            LocalProcessBackendConfig, ContainerBackendConfig,
+                            or None to use the backend's default config class.
+                            Defaults to KubernetesBackendConfig.
 
         Raises:
             ValueError: Invalid backend configuration.
@@ -52,6 +59,8 @@ def __init__(
             self.backend = KubernetesBackend(backend_config)
         elif isinstance(backend_config, LocalProcessBackendConfig):
             self.backend = LocalProcessBackend(backend_config)
+        elif isinstance(backend_config, ContainerBackendConfig):
+            self.backend = ContainerBackend(backend_config)
         else:
             raise ValueError(f"Invalid backend config '{backend_config}'")
 

kubeflow/trainer/backends/container/adapters/base.py

@@ -0,0 +1,195 @@
+# Copyright 2025 The Kubeflow Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Container client adapters for Docker and Podman.
+
+This module implements the adapter pattern to abstract away differences between
+Docker and Podman APIs, allowing the backend to work with either runtime through
+a common interface.
+"""
+
+from __future__ import annotations
+
+import abc
+from collections.abc import Iterator
+from typing import Optional
+
+
+class BaseContainerClientAdapter(abc.ABC):
+    """
+    Abstract adapter interface for container clients.
+
+    This adapter abstracts the container runtime API, allowing the backend
+    to work with Docker and Podman through a unified interface.
+    """
+
+    @abc.abstractmethod
+    def ping(self):
+        """Test the connection to the container runtime."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def create_network(
+        self,
+        name: str,
+        labels: dict[str, str],
+    ) -> str:
+        """
+        Create a container network.
+
+        Args:
+            name: Network name
+            labels: Labels to attach to the network
+
+        Returns:
+            Network ID or name
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def delete_network(self, network_id: str):
+        """Delete a network."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def create_and_start_container(
+        self,
+        image: str,
+        command: list[str],
+        name: str,
+        network_id: str,
+        environment: dict[str, str],
+        labels: dict[str, str],
+        volumes: dict[str, dict[str, str]],
+        working_dir: str,
+    ) -> str:
+        """
+        Create and start a container.
+
+        Args:
+            image: Container image
+            command: Command to run
+            name: Container name
+            network_id: Network to attach to
+            environment: Environment variables
+            labels: Container labels
+            volumes: Volume mounts
+            working_dir: Working directory
+
+        Returns:
+            Container ID
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def get_container(self, container_id: str):
+        """Get container object by ID."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def container_logs(self, container_id: str, follow: bool) -> Iterator[str]:
+        """Stream logs from a container."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def stop_container(self, container_id: str, timeout: int = 10):
+        """Stop a container."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def remove_container(self, container_id: str, force: bool = True):
+        """Remove a container."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def pull_image(self, image: str):
+        """Pull an image."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def image_exists(self, image: str) -> bool:
+        """Check if an image exists locally."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def run_oneoff_container(self, image: str, command: list[str]) -> str:
+        """
+        Run a short-lived container and return its output.
+
+        Args:
+            image: Container image
+            command: Command to run
+
+        Returns:
+            Container output as string
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def container_status(self, container_id: str) -> tuple[str, Optional[int]]:
+        """
+        Get container status.
+
+        Returns:
+            Tuple of (status_string, exit_code)
+            Status strings: "running", "created", "exited", etc.
+            Exit code is None if container hasn't exited
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def get_container_ip(self, container_id: str, network_id: str) -> Optional[str]:
+        """
+        Get container's IP address on a specific network.
+
+        Args:
+            container_id: Container ID
+            network_id: Network name or ID
+
+        Returns:
+            IP address string or None if not found
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def list_containers(self, filters: Optional[dict[str, list[str]]] = None) -> list[dict]:
+        """
+        List containers, optionally filtered by labels.
+
+        Args:
+            filters: Dictionary of filters (e.g., {"label": ["key=value"]})
+
+        Returns:
+            List of container info dictionaries with keys:
+            - id: Container ID
+            - name: Container name
+            - labels: Dictionary of labels
+            - status: Container status
+            - created: Creation timestamp
+        """
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def get_network(self, network_id: str) -> Optional[dict]:
+        """
+        Get network information by ID or name.
+
+        Args:
+            network_id: Network ID or name
+
+        Returns:
+            Dictionary with network info including labels, or None if not found
+        """
+        raise NotImplementedError()