Add docker context create command (#394)

docs/template/sub-commands/context.md

@@ -0,0 +1,40 @@
+## How to use Docker contexts
+
+Docker contexts allow you to connect to docker daemons other than the local one. This is similar to the `-H` argument of the Docker command.
+
+Contexts commands allow you to declare, save, list local and remote Docker daemons and Kubernetes endpoints that you have.
+
+An exemple here with python-on-whales:
+
+```python
+from python_on_whales import docker, DockerContextConfig
+
+new_context = docker.context.create(
+    "my_remote_ssh_server",
+    docker=DockerContextConfig(host="ssh://[email protected]"),
+    description="my server ssh with a lot more power"
+)
+print(docker.context.list())
+# [python_on_whales.Context(name='default', endpoints={'docker': ContextEndpoint(host='unix:///var/run/docker.sock', skip_tls_verify=False)}),
+# python_on_whales.Context(name='my_remote_ssh_server', endpoints={'docker': ContextEndpoint(host='ssh://[email protected]', skip_tls_verify=False)})]
+new_context.use()
+# it's the same to use docker.context.use("my_remote_ssh_server") or docker.context.use(new_context)
+
+print(docker.ps()) # will list the containers in the remote server
+# [python_on_whales.Container(id=...), python_on_whales.Container(id=...)]
+# return to the local docker daemon
+docker.context.use("default")
+print(docker.ps()) # will list the containers running locally
+# [python_on_whales.Container(id=...)]
+```
+
+Note that for this simple use case, it's equivalent to use the `-H` option of the Docker client like so:
+```python
+from python_on_whales import DockerClient
+
+docker = DockerClient(host="ssh://[email protected]")
+
+print(docker.ps())
+```
+
+{{autogenerated}}

python_on_whales/__init__.py

@@ -2,7 +2,11 @@
 from .components.buildx.cli_wrapper import Builder
 from .components.config.cli_wrapper import Config
 from .components.container.cli_wrapper import Container, ContainerStats
-from .components.context.cli_wrapper import Context
+from .components.context.cli_wrapper import (
+    Context,
+    DockerContextConfig,
+    KubernetesContextConfig,
+)
 from .components.image.cli_wrapper import Image
 from .components.network.cli_wrapper import Network
 from .components.node.cli_wrapper import Node

python_on_whales/components/context/cli_wrapper.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Union, overload
 
 from python_on_whales.client_config import (
@@ -76,14 +78,102 @@ def use(self) -> None:
         """Use this context"""
         ContextCLI(self.client_config).use(self)
 
+    def __repr__(self):
+        return (
+            f"python_on_whales.Context(name='{self.name}', endpoints={self.endpoints})"
+        )
+
 
 ValidContext = Union[Context, str]
 
 
+@dataclass
+class DockerContextConfig:
+    from_: Optional[ValidContext] = None
+    host: Optional[str] = None
+    certificate_authority: Union[Path, str, None] = None
+    certificate: Union[Path, str, None] = None
+    key: Union[Path, str, None] = None
+    skip_tls_verify: bool = False
+
+    def format_for_docker_cli(self) -> str:
+        list_of_args = []
+        if self.from_ is not None:
+            list_of_args.append(f"from={self.from_}")
+        if self.host is not None:
+            list_of_args.append(f"host={self.host}")
+        if self.certificate_authority is not None:
+            list_of_args.append(f"ca={self.certificate_authority}")
+        if self.certificate is not None:
+            list_of_args.append(f"cert={self.certificate}")
+        if self.key is not None:
+            list_of_args.append(f"key={self.key}")
+        list_of_args.append(f"skip-tls-verify={self.skip_tls_verify}")
+
+        return ",".join(list_of_args)
+
+
+@dataclass
+class KubernetesContextConfig:
+    from_: Optional[ValidContext] = None
+    config_file: Union[str, Path, None] = None
+    context_override: Optional[str] = None
+    namespace_override: Optional[str] = None
+
+    def format_for_docker_cli(self) -> str:
+        list_of_args = []
+        if self.from_ is not None:
+            list_of_args.append(f"from={self.from_}")
+        if self.config_file is not None:
+            list_of_args.append(f"config-file={self.config_file}")
+        if self.context_override is not None:
+            list_of_args.append(f"context-override={self.context_override}")
+        if self.namespace_override is not None:
+            list_of_args.append(f"namespace-override={self.namespace_override}")
+        return ",".join(list_of_args)
+
+
 class ContextCLI(DockerCLICaller):
-    def create(self):
-        """Not yet implemented"""
-        raise NotImplementedError
+    def create(
+        self,
+        context_name: str,
+        default_stack_orchestrator: Optional[str] = None,
+        description: Optional[str] = None,
+        from_: Optional[ValidContext] = None,
+        docker: Union[Dict[str, Any], DockerContextConfig, None] = None,
+        kubernetes: Union[Dict[str, Any], KubernetesContextConfig, None] = None,
+    ) -> Context:
+        """Creates a new context
+
+        # Arguments
+            context: name of the context to create
+            default_stack_orchestrator: Default orchestrator for stack operations to use with this context (swarm|kubernetes|all)
+            description: Description of the context
+            docker: Set the docker endpoint, you can use a dict of a class to
+                specify the options. The class is `python_on_whales.DockerContextConfig`.
+            from_: Create context from a named context
+            kubernetes: Set the kubernetes endpoint. You can use a dict or a class to specify the options. The class
+                is `python_on_whales.KubernetesContextConfig`.
+        """
+        if isinstance(docker, dict):
+            docker = DockerContextConfig(**docker)
+        if isinstance(kubernetes, dict):
+            kubernetes = KubernetesContextConfig(**kubernetes)
+
+        full_cmd = self.docker_cmd + ["context", "create"]
+
+        full_cmd.add_simple_arg(
+            "--default-stack-orchestrator", default_stack_orchestrator
+        )
+        full_cmd.add_simple_arg("--description", description)
+        full_cmd.add_simple_arg("--from", from_)
+        if docker is not None:
+            full_cmd.add_simple_arg("--docker", docker.format_for_docker_cli())
+        if kubernetes is not None:
+            full_cmd.add_simple_arg("--kubernetes", kubernetes.format_for_docker_cli())
+        full_cmd.append(context_name)
+        run(full_cmd)
+        return self.inspect(context_name)
 
     @overload
     def inspect(self, x: Union[None, str]) -> Context:

tests/python_on_whales/components/test_context.py

@@ -12,6 +12,24 @@ def test_load_json(json_file):
     # we could do more checks here if needed
 
 
+def test_create_context():
+    testname = "testpow"
+    host = "ssh://[email protected]"
+    description = "Python on whales testing context"
+
+    all_contexts_before = set(docker.context.list())
+    with docker.context.create(
+        testname, docker=dict(host=host), description=description
+    ) as new_context:
+        assert new_context.name == testname
+        assert new_context.endpoints["docker"].host == host
+        assert new_context.metadata["Description"] == description
+
+        assert new_context not in all_contexts_before
+
+        assert new_context in docker.context.list()
+
+
 def test_inpect():
     default_context = docker.context.inspect()
     assert default_context.name == "default"