"""Client for connecting to Podman service."""
import logging
import os
from contextlib import AbstractContextManager
from pathlib import Path
from typing import Any, Dict, Optional
import xdg.BaseDirectory
from podman.api import cached_property
from podman.api.client import APIClient
from podman.domain.config import PodmanConfig
from podman.domain.containers_manager import ContainersManager
from podman.domain.events import EventsManager
from podman.domain.images_manager import ImagesManager
from podman.domain.manifests import ManifestsManager
from podman.domain.networks_manager import NetworksManager
from podman.domain.pods_manager import PodsManager
from podman.domain.secrets import SecretsManager
from podman.domain.system import SystemManager
from podman.domain.volumes import VolumesManager
logger = logging.getLogger("podman")
[docs]
class PodmanClient(AbstractContextManager):
"""Client to connect to a Podman service.
Examples:
with PodmanClient(base_url="ssh://root@api.example:22/run/podman/podman.sock?secure=True",
identity="~alice/.ssh/api_ed25519")
"""
def __init__(self, **kwargs) -> None:
"""Initialize PodmanClient.
Keyword Args:
base_url (str): Full URL to Podman service. See examples.
version (str): API version to use. Default: auto, use version from server
timeout (int): Timeout for API calls, in seconds.
Default: socket._GLOBAL_DEFAULT_TIMEOUT.
tls: Ignored. SSH connection configuration delegated to SSH Host configuration.
user_agent (str): User agent for service connections. Default: PodmanPy/<Code Version>
credstore_env (Mapping[str, str]): Dict containing environment for credential store
use_ssh_client (True): Always shell out to SSH client for
SSH Podman service connections.
max_pool_size (int): Number of connections to save in pool
connection (str): Identifier of connection to use from
XDG_CONFIG_HOME/containers/containers.conf
identity (str): Provide SSH key to authenticate SSH connection.
Examples:
base_url:
- http+ssh://<user>@<host>[:port]</run/podman/podman.sock>[?secure=True]
- http+unix://</run/podman/podman.sock>
- tcp://<localhost>[:<port>]
"""
super().__init__()
config = PodmanConfig()
api_kwargs = kwargs.copy()
if "connection" in api_kwargs:
connection = config.services[api_kwargs.get("connection")]
api_kwargs["base_url"] = connection.url.geturl()
# Override configured identity, if provided in arguments
api_kwargs["identity"] = kwargs.get("identity", str(connection.identity))
elif "base_url" not in api_kwargs:
path = str(
Path(xdg.BaseDirectory.get_runtime_dir(strict=False)) / "podman" / "podman.sock"
)
api_kwargs["base_url"] = "http+unix://" + path
self.api = APIClient(**api_kwargs)
def __enter__(self) -> "PodmanClient":
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.close()
[docs]
@classmethod
def from_env(
cls,
version: str = "auto",
timeout: Optional[int] = None,
max_pool_size: Optional[int] = None,
ssl_version: Optional[int] = None, # pylint: disable=unused-argument
assert_hostname: bool = False, # pylint: disable=unused-argument
environment: Optional[Dict[str, str]] = None,
credstore_env: Optional[Dict[str, str]] = None,
use_ssh_client: bool = True, # pylint: disable=unused-argument
) -> "PodmanClient":
"""Returns connection to service using environment variables and parameters.
Environment variables:
- DOCKER_HOST, CONTAINER_HOST: URL to Podman service
- DOCKER_TLS_VERIFY, CONTAINER_TLS_VERIFY: Verify host against CA certificate
- DOCKER_CERT_PATH, CONTAINER_CERT_PATH: Path to TLS certificates for host connection
Args:
version: API version to use. Default: auto, use version from server
timeout: Timeout for API calls, in seconds.
max_pool_size: Number of connections to save in pool.
ssl_version: SSH configuration delegated to SSH client configuration. Ignored.
assert_hostname: Ignored.
environment: Dict containing input environment. Default: os.environ
credstore_env: Dict containing environment for credential store
use_ssh_client: Use system ssh client rather than ssh module. Always, True.
Returns:
Client used to communicate with a Podman service.
Raises:
ValueError when required environment variable is not set
"""
environment = environment or os.environ
credstore_env = credstore_env or {}
if version == "auto":
version = None
host = environment.get("CONTAINER_HOST") or environment.get("DOCKER_HOST") or None
if host is None:
raise ValueError("CONTAINER_HOST or DOCKER_HOST must be set to URL of podman service.")
return PodmanClient(
base_url=host,
version=version,
timeout=timeout,
tls=False,
credstore_env=credstore_env,
max_pool_size=max_pool_size,
)
@cached_property
def containers(self) -> ContainersManager:
"""Returns Manager for operations on containers stored by a Podman service."""
return ContainersManager(client=self.api)
@cached_property
def images(self) -> ImagesManager:
"""Returns Manager for operations on images stored by a Podman service."""
return ImagesManager(client=self.api)
@cached_property
def manifests(self) -> ManifestsManager:
"""Returns Manager for operations on manifests maintained by a Podman service."""
return ManifestsManager(client=self.api)
@cached_property
def networks(self) -> NetworksManager:
"""Returns Manager for operations on networks maintained by a Podman service."""
return NetworksManager(client=self.api)
@cached_property
def volumes(self) -> VolumesManager:
"""Returns Manager for operations on volumes maintained by a Podman service."""
return VolumesManager(client=self.api)
@cached_property
def pods(self) -> PodsManager:
"""Returns Manager for operations on pods maintained by a Podman service."""
return PodsManager(client=self.api)
@cached_property
def secrets(self):
"""Returns Manager for operations on secrets maintained by a Podman service."""
return SecretsManager(client=self.api)
@cached_property
def system(self):
return SystemManager(client=self.api)
[docs]
def df(self) -> Dict[str, Any]: # pylint: disable=missing-function-docstring,invalid-name
return self.system.df()
df.__doc__ = SystemManager.df.__doc__
[docs]
def events(self, *args, **kwargs): # pylint: disable=missing-function-docstring
return EventsManager(client=self.api).list(*args, **kwargs)
events.__doc__ = EventsManager.list.__doc__
[docs]
def info(self, *args, **kwargs): # pylint: disable=missing-function-docstring
return self.system.info(*args, **kwargs)
info.__doc__ = SystemManager.info.__doc__
[docs]
def login(self, *args, **kwargs): # pylint: disable=missing-function-docstring
return self.system.login(*args, **kwargs)
login.__doc__ = SystemManager.login.__doc__
[docs]
def ping(self) -> bool: # pylint: disable=missing-function-docstring
return self.system.ping()
ping.__doc__ = SystemManager.ping.__doc__
[docs]
def version(self, *args, **kwargs): # pylint: disable=missing-function-docstring
_ = args
return self.system.version(**kwargs)
version.__doc__ = SystemManager.version.__doc__
[docs]
def close(self):
"""Release PodmanClient Resources."""
return self.api.close()
@property
def swarm(self):
"""Swarm not supported.
Raises:
NotImplemented: Swarm not supported by Podman service
"""
raise NotImplementedError("Swarm operations are not supported by Podman service.")
# Aliases to cover all swarm methods
services = swarm
configs = swarm
nodes = swarm
# Aliases to minimize effort to port to PodmanPy
DockerClient = PodmanClient
from_env = PodmanClient.from_env
