"""Asynchronous Python client for Python Portainer.""" from __future__ import annotations import asyncio import json import socket from dataclasses import dataclass from datetime import UTC, datetime, timedelta from importlib import metadata from typing import TYPE_CHECKING, Any, Self from urllib.parse import urlparse from aiohttp import ClientError, ClientResponseError, ClientSession from aiohttp.hdrs import METH_DELETE, METH_GET, METH_POST from tenacity import AsyncRetrying, retry_if_exception_type, stop_after_attempt, wait_exponential from yarl import URL from pyportainer.exceptions import ( PortainerAuthenticationError, PortainerConnectionError, PortainerError, PortainerNotFoundError, PortainerTimeoutError, ) from pyportainer.models.docker import ( DockerContainer, DockerContainerCPUStats, DockerContainerStats, DockerDFType, DockerEvent, DockerImagePruneResponse, DockerSystemDF, DockerVolume, ImageInformation, LocalImageInformation, PortainerImageUpdateStatus, ) from pyportainer.models.docker_inspect import DockerInfo, DockerInspect, DockerVersion from pyportainer.models.portainer import Endpoint, PortainerSystemStatus from pyportainer.models.stacks import Stack if TYPE_CHECKING: from collections.abc import AsyncGenerator try: VERSION = metadata.version(__package__) except metadata.PackageNotFoundError: # pragma: no cover VERSION = "DEV-0.0.0" # pylint: disable=invalid-name @dataclass class Portainer: """Main class for handling connections with the Python Portainer API.""" request_timeout: float = 10.0 session: ClientSession | None = None _close_session: bool = False def __init__( # pylint: disable=too-many-arguments self, api_url: str, api_key: str, *, request_timeout: float = 10.0, session: ClientSession | None = None, max_retries: int = 3, ) -> None: """Initialize the Portainer object. Args: ---- api_url: URL of the Portainer API. api_key: API key for authentication. request_timeout: Timeout for requests (in seconds). session: Optional aiohttp session to use. max_retries: Maximum number of retry attempts on transient errors. """ self._api_key = api_key self._request_timeout = request_timeout self._session = session self._max_retries = max_retries parsed_url = urlparse(api_url) self._api_host = parsed_url.hostname or "" self._api_scheme = parsed_url.scheme or "" self._api_port = parsed_url.port self._api_base_path = (parsed_url.path or "").rstrip("/") self._prev_container_stats: dict[tuple[int, str], DockerContainerStats] | None = None # pylint: disable=too-many-arguments, too-many-locals, too-many-branches async def _request( self, uri: str, *, method: str = METH_GET, params: dict[str, Any] | None = None, json_body: dict[str, Any] | None = None, timeout: float | None = None, parse: bool = True, ) -> Any: """Handle a request to the Python Portainer API. Args: ---- uri: Request URI, without '/api/', for example, 'status'. method: HTTP method to use. params: Extra options to improve or limit the response. timeout: Timeout for the request (in seconds). parse: Whether to parse the response as JSON. Returns: ------- A Python dictionary (JSON decoded) with the response from the Python Portainer API. Raises: ------ Python PortainerAuthenticationError: If the API key is invalid. """ url = URL.build( scheme=self._api_scheme, host=self._api_host, port=self._api_port, path=f"{self._api_base_path}/api/", ).join(URL(uri)) headers = { "Accept": "application/json, text/plain", "User-Agent": f"PythonPortainer/{VERSION}", "X-API-Key": self._api_key, } if self._session is None: self._session = ClientSession() self._close_session = True # Only override timeout if a specific value is provided, else use default if timeout is None: timeout = self._request_timeout async for attempt in AsyncRetrying( retry=retry_if_exception_type((PortainerConnectionError, PortainerTimeoutError)), wait=wait_exponential(multiplier=1, min=1, max=10), stop=stop_after_attempt(self._max_retries + 1), reraise=True, ): with attempt: try: async with asyncio.timeout(timeout): response = await self._session.request( method, url, headers=headers, params=params, json=json_body, ) response.raise_for_status() except TimeoutError as err: msg = f"Timeout error while accessing {method} {url}: {err}" raise PortainerTimeoutError(msg) from err except ClientResponseError as err: match err.status: case 401: msg = f"Authentication failed for {method} {url}: Invalid API key" raise PortainerAuthenticationError(msg) from err case 404: msg = f"Resource not found at {method} {url}: {err}" raise PortainerNotFoundError(msg) from err case _: msg = f"Connection error for {method} {url}: {err}" raise PortainerConnectionError(msg) from err except (ClientError, socket.gaierror) as err: msg = f"Unexpected error during {method} {url}: {err}" raise PortainerConnectionError(msg) from err if response.status in (204, 304): return None content_type = response.headers.get("Content-Type", "") if "application/json" not in content_type: text = await response.text() msg = "Unexpected content type response from the Portainer API" raise PortainerError( msg, {"Content-Type": content_type, "response": text}, ) # Read events instead. Ideal for getting image pull progress events: list[Any] = [] if not parse: async for chunk in response.content: for line in chunk.splitlines(): stripped_line = line.strip() if not stripped_line: continue events.append(json.loads(stripped_line)) return events return await response.json() async def _stream_request( self, uri: str, *, params: dict[str, Any] | None = None, ) -> AsyncGenerator[dict[str, Any], None]: """Open a persistent streaming connection and yield JSON events as they arrive. Unlike :meth:`_request`, this method does not buffer the full response. The connection remains open until cancelled or the server closes it. The connection-establishment step is subject to the normal request timeout; the ongoing stream is not time-limited. Args: ---- uri: Request URI, without '/api/'. params: Query parameters to include in the request. Yields: ------ Parsed JSON objects, one per newline-delimited event. Raises: ------ PortainerTimeoutError: If the connection cannot be established within the timeout. PortainerAuthenticationError: If the API key is invalid. PortainerConnectionError: On network errors. """ url = URL.build( scheme=self._api_scheme, host=self._api_host, port=self._api_port, path=f"{self._api_base_path}/api/", ).join(URL(uri)) headers = { "Accept": "application/json, text/plain", "User-Agent": f"PythonPortainer/{VERSION}", "X-API-Key": self._api_key, } if self._session is None: self._session = ClientSession() self._close_session = True try: async with asyncio.timeout(self._request_timeout): response = await self._session.request( METH_GET, url, headers=headers, params=params, ) response.raise_for_status() except TimeoutError as err: msg = f"Timeout error while connecting to {url}: {err}" raise PortainerTimeoutError(msg) from err except ClientResponseError as err: match err.status: case 401: msg = f"Authentication failed for {url}: Invalid API key" raise PortainerAuthenticationError(msg) from err case 404: msg = f"Resource not found at {url}: {err}" raise PortainerNotFoundError(msg) from err case _: msg = f"Connection error for {url}: {err}" raise PortainerConnectionError(msg) from err except (ClientError, socket.gaierror) as err: msg = f"Unexpected error connecting to {url}: {err}" raise PortainerConnectionError(msg) from err try: buffer = b"" async for chunk in response.content: buffer += chunk while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) stripped = line.strip() if stripped: yield json.loads(stripped) finally: response.release() async def get_events( self, endpoint_id: int, *, since: datetime | None = None, until: datetime | None = None, filters: dict[str, list[str]] | None = None, ) -> AsyncGenerator[DockerEvent, None]: """Stream Docker events from an endpoint in real time. Opens a persistent connection to the Docker events endpoint and yields :class:`~pyportainer.models.docker.DockerEvent` objects as they are emitted. When ``until`` is provided the Docker daemon closes the connection once all matching events have been sent, making this suitable for bounded queries as well as infinite streams. Args: ---- endpoint_id: The ID of the Portainer endpoint to stream events from. since: Only return events after this timestamp. If timezone-naive, UTC is assumed. until: Only return events before this timestamp. If timezone-naive, UTC is assumed. When supplied, the stream ends automatically. filters: Optional Docker event filters, e.g. ``{"type": ["container"], "event": ["start", "die"]}``. Yields: ------ :class:`~pyportainer.models.docker.DockerEvent` objects. """ params: dict[str, Any] = {} if since is not None: params["since"] = int(since.timestamp()) if until is not None: params["until"] = int(until.timestamp()) if filters is not None: params["filters"] = json.dumps(filters) async for raw in self._stream_request( f"endpoints/{endpoint_id}/docker/events", params=params or None, ): yield DockerEvent.from_dict(raw) async def get_recent_events( self, endpoint_id: int, *, since: datetime, until: datetime | None = None, filters: dict[str, list[str]] | None = None, ) -> list[DockerEvent]: """Return Docker events from an endpoint for a bounded time window. Unlike :meth:`get_events`, this method collects all matching events into a list and returns once the time window is exhausted. It is a thin wrapper around :meth:`get_events` that passes ``until`` (defaulting to now) so the Docker daemon closes the connection automatically. Args: ---- endpoint_id: The ID of the Portainer endpoint to query. since: Only return events after this timestamp. If timezone-naive, UTC is assumed. until: Only return events before this timestamp. Defaults to the current UTC time if not provided. filters: Optional Docker event filters, e.g. ``{"type": ["container"], "event": ["start", "die"]}``. Returns: ------- A list of :class:`~pyportainer.models.docker.DockerEvent` objects, ordered as received from the Docker daemon. """ if until is None: until = datetime.now(UTC) return [ event async for event in self.get_events( endpoint_id, since=since, until=until, filters=filters, ) ] async def get_endpoints(self) -> list[Endpoint]: """Get the list of endpoints from the Portainer API. Returns ------- A list of Endpoint objects. """ endpoints = await self._request("endpoints") return [Endpoint.from_dict(endpoint) for endpoint in endpoints] async def get_containers(self, endpoint_id: int) -> list[DockerContainer]: """Get the list of containers from the Portainer API. Args: ---- endpoint_id: The ID of the endpoint to get containers from. all: If True, include all containers. If False, only running containers. Returns: ------- A list of containers. """ containers = await self._request(f"endpoints/{endpoint_id}/docker/containers/json?all=1") return [DockerContainer.from_dict(container) for container in containers] async def start_container(self, endpoint_id: int, container_id: str) -> Any: """Start a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to start. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/start", method="POST", json_body={}, ) async def stop_container(self, endpoint_id: int, container_id: str) -> Any: """Stop a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to stop. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/stop", method="POST", ) async def restart_container(self, endpoint_id: int, container_id: str) -> Any: """Restart a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to restart. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/restart", method="POST", ) async def pause_container(self, endpoint_id: int, container_id: str) -> Any: """Pause a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to pause. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/pause", method="POST", ) async def unpause_container(self, endpoint_id: int, container_id: str) -> Any: """Unpause a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to unpause. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/unpause", method="POST", ) async def kill_container(self, endpoint_id: int, container_id: str) -> Any: """Kill a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to kill. """ return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/kill", method="POST", ) async def delete_container(self, endpoint_id: int, container_id: str, *, force: bool = False) -> Any: """Delete a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to delete. force: If True, force delete the container. """ params = {"force": str(force).lower()} return await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}", method="DELETE", params=params, ) async def inspect_container(self, endpoint_id: int, container_id: str, *, raw: bool = False) -> DockerInspect | Any: """Inspect a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to inspect. raw: If True, return the raw JSON response. If False, return a DockerInspect object. Returns: ------- A DockerContainer object with the inspected data. """ container = await self._request(f"endpoints/{endpoint_id}/docker/containers/{container_id}/json") if raw: return container return DockerInspect.from_dict(container) async def docker_version(self, endpoint_id: int) -> DockerVersion: """Get the Docker version on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. Returns: ------- A DockerVersion object with the Docker version data. """ version = await self._request(f"endpoints/{endpoint_id}/docker/version") return DockerVersion.from_dict(version) async def docker_info(self, endpoint_id: int) -> DockerInfo: """Get the Docker info on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. Returns: ------- A DockerInfo object with the Docker info data. """ info = await self._request(f"endpoints/{endpoint_id}/docker/info") return DockerInfo.from_dict(info) async def container_stats( self, endpoint_id: int, container_id: str, *, stream: bool = False, one_shot: bool = True, ) -> Any: """Get the stats of a container on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to get stats from. stream: If True, stream the stats. If False, get a single snapshot. one_shot: If True, get a single snapshot. If False, stream the stats. Returns: ------- The stats of the container. """ params = {"stream": str(stream).lower(), "one-shot": str(one_shot).lower()} stats = await self._request( f"endpoints/{endpoint_id}/docker/containers/{container_id}/stats", params=params, ) return DockerContainerStats.from_dict(stats) async def get_image_information(self, endpoint_id: int, image_id: str) -> ImageInformation: """Get information about a Docker image. Args: ---- endpoint_id: The ID of the endpoint. image_id: The ID of the image to get information about. Returns: ------- An ImageInformation object with the image data. """ image = await self._request(f"endpoints/{endpoint_id}/docker/distribution/{image_id}/json") return ImageInformation.from_dict(image) async def get_image(self, endpoint_id: int, image_id: str) -> LocalImageInformation: """Get information about a Docker image. Args: ---- endpoint_id: The ID of the endpoint. image_id: The ID of the image to get information about. Returns: ------- A LocalImageInformation object with the image data. """ image = await self._request(f"endpoints/{endpoint_id}/docker/images/{image_id}/json") return LocalImageInformation.from_dict(image) async def container_image_status(self, endpoint_id: int, image: str) -> PortainerImageUpdateStatus: """Check whether a newer version of a Docker image is available in the registry. Args: ---- endpoint_id: The ID of the endpoint. image: The image name (with optional tag) to check. Returns: ------- A PortainerImageUpdateStatus with the comparison result and digests. """ local, remote = await asyncio.gather( self.get_image(endpoint_id, image), self.get_image_information(endpoint_id, image), ) registry_digest = remote.descriptor.digest if remote.descriptor else None local_digest = next( (digest.partition("@")[2] for digest in (local.repo_digests or []) if "@" in digest), None, ) return PortainerImageUpdateStatus( update_available=bool(registry_digest and registry_digest != local_digest), local_digest=local_digest, registry_digest=registry_digest, ) async def image_recreate(self, endpoint_id: int, image_id: str, timeout: timedelta = timedelta(minutes=5)) -> Any: """Recreate a Docker image. Args: ---- endpoint_id: The ID of the endpoint. image_id: The ID of the image to recreate. timeout: Timeout for the image recreation process. Defaults to 5 minutes. Returns: ------- An ImageInformation object with the recreated image data. """ params = {"fromImage": image_id} return await self._request( uri=f"endpoints/{endpoint_id}/docker/images/create?fromImage={image_id}", timeout=timeout.total_seconds(), method="POST", params=params, parse=False, ) async def container_recreate_helper(self, endpoint_id: int, container_id: str, image: str, timeout: timedelta = timedelta(minutes=5)) -> Any: """Recreate a Docker container service. This helper runs through the Portainer API and recreates the specified container. It first inspects the container to get its configuration, then creates a new container with the same configuration. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to recreate. image: The tag of the image to use for the new container. Returns: ------- The response from the Portainer API. """ container_inspect = await self.inspect_container( endpoint_id=endpoint_id, container_id=container_id, raw=True, ) if not isinstance(container_inspect, dict): msg = "Failed to inspect container for recreation." raise PortainerError(msg) await self.image_recreate( endpoint_id=endpoint_id, image_id=image, timeout=timeout, ) await self.stop_container( endpoint_id=endpoint_id, container_id=container_id, ) await self.delete_container( endpoint_id=endpoint_id, container_id=container_id, force=True, ) create_body = { **container_inspect["Config"], "Image": image, "HostConfig": container_inspect["HostConfig"], "Config": container_inspect["Config"], } created = await self.container_create( endpoint_id=endpoint_id, name=container_inspect.get("Name", "").lstrip("/"), image=image, config=create_body, ) # This is optional; reattach networks the same way as the original container # I have to test this, probablt need some friendly users... networks = (container_inspect["NetworkSettings"] or {}).get("Networks") or {} for net_name in networks: network_name = (container_inspect["HostConfig"] or {}).get("NetworkMode") or "" if network_name in {"host", "none"} or network_name.startswith("container:"): continue await self._request( f"endpoints/{endpoint_id}/docker/networks/{net_name}/connect", method="POST", json_body={"Container": created.id}, ) await self.start_container( endpoint_id=endpoint_id, container_id=created.id, ) return created async def container_recreate( self, endpoint_id: int, container_id: str, timeout: timedelta = timedelta(minutes=5), *, pull_image: bool = False ) -> DockerContainer: """Recreate a Docker container. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container to recreate. timeout: Timeout for the container recreation process. Defaults to 5 minutes. pull_image: If True, pull the latest image before recreating the container. Returns: ------- The response from the Portainer API. """ params = {"PullImage": pull_image} container = await self._request( uri=f"docker/{endpoint_id}/containers/{container_id}/recreate", method="POST", json_body=params, timeout=timeout.total_seconds(), ) if isinstance(container.get("State"), dict): container["State"] = container["State"].get("Status") return DockerContainer.from_dict(container) async def container_create(self, endpoint_id: int, name: str, image: str, config: dict[str, Any]) -> DockerContainer: """Create a Docker container. Args: ---- endpoint_id: The ID of the endpoint. name: The name of the container to create. Returns: ------- A DockerContainer object with the created container data. """ params = {"name": name} json_body = {"Image": image} json_body.update(config) container = await self._request( uri=f"endpoints/{endpoint_id}/docker/containers/create", method="POST", params=params, json_body=json_body, ) return DockerContainer.from_dict(container) async def images_prune(self, endpoint_id: int, until: timedelta | None, *, dangling: bool) -> Any: """Prune Docker images on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. dangling: When set to true (or 1), prune only unused and untagged images. When set to false (or 0), all unused images are pruned. until: Prune images created before this timestamp. The until is a timedelta that specifies the duration before the current time. The can be Unix timestamps, date formatted timestamps, or Go duration strings (e.g. 10m, 1h30m). Returns: ------- The response from the Portainer API. """ params: dict[str, Any] = {"dangling": str(dangling).lower()} if until is not None: params["until"] = int((datetime.now(UTC) - until).timestamp()) response = await self._request( f"endpoints/{endpoint_id}/docker/images/prune", method="POST", params=params, ) return DockerImagePruneResponse.from_dict(response) async def docker_system_df(self, endpoint_id: int, data_type: DockerDFType | None = None, *, verbose: bool = False) -> Any: """Get Docker system disk usage on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. data_type: The type of resource to filter by. Use DockerDFType enum values. verbose: If True, include detailed information. Returns: ------- The response from the Portainer API. """ params: dict[str, Any] = {"verbose": str(verbose).lower()} if data_type is not None: params["type"] = data_type response = await self._request( f"endpoints/{endpoint_id}/docker/system/df", method="GET", params=params, ) return DockerSystemDF.from_dict(response) async def portainer_system_status(self) -> PortainerSystemStatus: """Get the system status of the Portainer instance. Returns ------- A PortainerSystemStatus object with the system status data. """ status = await self._request("system/status") return PortainerSystemStatus.from_dict(status) async def get_stacks( self, endpoint_id: int | None = None, swarm_id: str | None = None, ) -> list[Stack]: """Get the list of stacks from the Portainer API. Args: ---- endpoint_id: Filter stacks by endpoint ID. swarm_id: Filter stacks by Swarm cluster ID. Returns: ------- A list of stacks. """ filters: dict[str, Any] = {} if endpoint_id is not None: filters["EndpointID"] = endpoint_id if swarm_id is not None: filters["SwarmID"] = swarm_id params = filters and {"filters": json.dumps(filters)} stacks = await self._request("stacks", params=params) if stacks is None: # 204 response = no stacks return [] return [Stack.from_dict(stack) for stack in stacks] async def get_stack(self, stack_id: int) -> Stack: """Get details of a specific stack. Args: ---- stack_id: The ID of the stack. Returns: ------- Stack details. """ stack = await self._request(f"stacks/{stack_id}") return Stack.from_dict(stack) async def get_stack_containers( self, endpoint_id: int, stack_name: str, ) -> list[DockerContainer]: """Get containers belonging to a stack. Filters containers by the com.docker.compose.project label. Args: ---- endpoint_id: The ID of the endpoint. stack_name: The name of the stack. Returns: ------- A list of containers in the stack. """ filters = {"label": [f"com.docker.compose.project={stack_name}"]} params = {"all": "1", "filters": json.dumps(filters)} containers = await self._request( f"endpoints/{endpoint_id}/docker/containers/json", params=params, ) return [DockerContainer.from_dict(container) for container in containers] async def start_stack(self, endpoint_id: int, stack_id: int, timeout: timedelta = timedelta(minutes=5)) -> Stack: """Start a stopped stack. Args: ---- stack_id: The ID of the stack. endpoint_id: The ID of the endpoint. timeout: The timeout for starting the stack. Returns: ------- Updated stack details. """ stack = await self._request( f"stacks/{stack_id}/start", method=METH_POST, params={"endpointId": endpoint_id}, timeout=timeout.total_seconds(), ) return Stack.from_dict(stack) async def stop_stack(self, endpoint_id: int, stack_id: int, timeout: timedelta = timedelta(minutes=5)) -> Stack: """Stop a running stack. Args: ---- stack_id: The ID of the stack. endpoint_id: The ID of the endpoint. timeout: The timeout for stopping the stack. Returns: ------- Updated stack details. """ stack = await self._request( f"stacks/{stack_id}/stop", method=METH_POST, params={"endpointId": endpoint_id}, timeout=timeout.total_seconds(), ) return Stack.from_dict(stack) async def delete_stack( self, stack_id: int, endpoint_id: int, *, external: bool = False, ) -> None: """Delete a stack. Args: ---- stack_id: The ID of the stack. endpoint_id: The ID of the endpoint. external: Set to True to delete an external Swarm stack. """ params: dict[str, Any] = { "endpointId": endpoint_id, "external": str(external).lower(), } await self._request( f"stacks/{stack_id}", method=METH_DELETE, params=params, ) async def get_volumes(self, endpoint_id: int) -> list[DockerVolume]: """Get the list of volumes from the Portainer API. Args: ---- endpoint_id: The ID of the endpoint to get volumes from. Returns: ------- A list of DockerVolume objects. """ volumes = await self._request(f"endpoints/{endpoint_id}/docker/volumes") return [DockerVolume.from_dict(volume) for volume in (volumes.get("Volumes") or [])] async def inspect_volume(self, endpoint_id: int, volume_name: str) -> DockerVolume: """Inspect a volume on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. volume_name: The name of the volume to inspect. Returns: ------- A DockerVolume object with the inspected volume data. """ volume = await self._request(f"endpoints/{endpoint_id}/docker/volumes/{volume_name}") return DockerVolume.from_dict(volume) async def prune_volumes(self, endpoint_id: int, *, all_volumes: bool = False) -> Any: """Prune unused volumes on the specified endpoint. Args: ---- endpoint_id: The ID of the endpoint. all_volumes: Set to True to prune all volumes, not just unused ones. Returns: ------- The response from the Portainer API. """ params = {"endpointId": endpoint_id, "all": str(all_volumes).lower()} return await self._request(f"endpoints/{endpoint_id}/docker/volumes/prune", method=METH_POST, params=params) async def get_container_cpu_usage(self, endpoint_id: int, container_id: str) -> DockerContainerCPUStats: """Get the current CPU usage percentage for the specified container. Args: ---- endpoint_id: The ID of the endpoint. container_id: The ID of the container. Returns: ------- The current CPU usage as a percentage. """ stats = await self.container_stats(endpoint_id, container_id, stream=False) docker_stats = DockerContainerCPUStats() num_cpus = len(stats.cpu_stats.cpu_usage.percpu_usage) if stats.cpu_stats.cpu_usage.percpu_usage else 1 if self._prev_container_stats is not None and (prev_stats := self._prev_container_stats.get((endpoint_id, container_id))): docker_stats.container_prev_stats = prev_stats cpu_delta = stats.cpu_stats.cpu_usage.total_usage - prev_stats.cpu_stats.cpu_usage.total_usage system_delta = stats.cpu_stats.system_cpu_usage - prev_stats.cpu_stats.system_cpu_usage cpu_kernel_delta = stats.cpu_stats.cpu_usage.usage_in_kernelmode - prev_stats.cpu_stats.cpu_usage.usage_in_kernelmode cpu_user_delta = stats.cpu_stats.cpu_usage.usage_in_usermode - prev_stats.cpu_stats.cpu_usage.usage_in_usermode if system_delta > 0: scale = num_cpus * 100.0 / system_delta if cpu_delta > 0: docker_stats.cpu_system_percentage = cpu_delta * scale if cpu_kernel_delta > 0: docker_stats.cpu_kernel_percentage = (cpu_kernel_delta + cpu_user_delta) * scale if cpu_user_delta > 0: docker_stats.cpu_user_percentage = (cpu_user_delta + cpu_kernel_delta) * scale docker_stats.cpu_system_usage = float(stats.cpu_stats.system_cpu_usage) docker_stats.online_cpus = stats.cpu_stats.online_cpus docker_stats.cpu_kernel_usage = float(stats.cpu_stats.cpu_usage.usage_in_kernelmode) docker_stats.cpu_user_usage = float(stats.cpu_stats.cpu_usage.usage_in_usermode) docker_stats.container_stats = stats self._prev_container_stats = self._prev_container_stats or {} self._prev_container_stats[(endpoint_id, container_id)] = stats return docker_stats async def close(self) -> None: """Close open client session.""" if self._session and self._close_session: await self._session.close() async def __aenter__(self) -> Self: """Async enter. Returns ------- The Portainer object. """ return self async def __aexit__(self, *_exc_info: object) -> None: """Async exit. Args: ---- _exc_info: Exec type. """ await self.close()