API Reference

Asynchronous Python client for Python Portainer.

DockerContainerState

Bases: StrEnum

Possible states of a Docker container.

Source code in src/pyportainer/models/docker.py

class DockerContainerState(StrEnum):
    """Possible states of a Docker container."""

    CREATED = "created"
    RESTARTING = "restarting"
    RUNNING = "running"
    REMOVING = "removing"
    PAUSED = "paused"
    EXITED = "exited"
    DEAD = "dead"

DockerDFType

Bases: StrEnum

Resource type filter for Docker system df.

Source code in src/pyportainer/models/docker.py

class DockerDFType(StrEnum):
    """Resource type filter for Docker system df."""

    CONTAINER = "container"
    IMAGE = "image"
    VOLUME = "volume"
    BUILD_CACHE = "build-cache"

DockerHealthStatus

Bases: StrEnum

Possible health states of a Docker container.

Source code in src/pyportainer/models/docker.py

class DockerHealthStatus(StrEnum):
    """Possible health states of a Docker container."""

    NONE = "none"
    STARTING = "starting"
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"

EndpointStatus

Bases: IntEnum

Portainer endpoint status.

Source code in src/pyportainer/models/docker.py

class EndpointStatus(IntEnum):
    """Portainer endpoint status."""

    UP = 1
    DOWN = 2

ImagePullProgress

Turn Docker image pull events into an overall percentage.

Docker announces every layer before the first download starts, but a layer's size is only known once its own download starts. Each layer therefore counts equally, the percentage stays 0 until a download starts and it never goes down.

Source code in src/pyportainer/image_pull.py

class ImagePullProgress:
    """Turn Docker image pull events into an overall percentage.

    Docker announces every layer before the first download starts, but a
    layer's size is only known once its own download starts. Each layer
    therefore counts equally, the percentage stays 0 until a download starts
    and it never goes down.
    """

    def __init__(self) -> None:
        """Initialize the progress tracker."""
        self._layers: dict[str, float] = {}
        self._started = False
        self._percentage = 0.0

    @property
    def percentage(self) -> float:
        """Return the overall progress of the pull, from 0 to 100."""
        return self._percentage

    def update(self, event: DockerImagePullEvent) -> float:
        """Process a pull event and return the overall percentage."""
        if event.id is None or (fraction := _layer_fraction(event)) is None:
            return self._percentage

        self._layers[event.id] = max(fraction, self._layers.get(event.id, 0.0))
        self._started = self._started or event.status not in _LAYER_START | {"Already exists"}
        if not self._started:
            return self._percentage

        self._percentage = max(
            self._percentage,
            sum(self._layers.values()) / len(self._layers) * 100,
        )
        return self._percentage

percentage property

Return the overall progress of the pull, from 0 to 100.

__init__()

Initialize the progress tracker.

Source code in src/pyportainer/image_pull.py

def __init__(self) -> None:
    """Initialize the progress tracker."""
    self._layers: dict[str, float] = {}
    self._started = False
    self._percentage = 0.0

update(event)

Process a pull event and return the overall percentage.

Source code in src/pyportainer/image_pull.py

def update(self, event: DockerImagePullEvent) -> float:
    """Process a pull event and return the overall percentage."""
    if event.id is None or (fraction := _layer_fraction(event)) is None:
        return self._percentage

    self._layers[event.id] = max(fraction, self._layers.get(event.id, 0.0))
    self._started = self._started or event.status not in _LAYER_START | {"Already exists"}
    if not self._started:
        return self._percentage

    self._percentage = max(
        self._percentage,
        sum(self._layers.values()) / len(self._layers) * 100,
    )
    return self._percentage

Portainer dataclass

Main class for handling connections with the Python Portainer API.

Source code in src/pyportainer/pyportainer.py

@dataclass
class Portainer:
    """Main class for handling connections with the Python Portainer API."""

    request_timeout: float = 10.0
    stream_timeout: float = 30.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,
        stream_timeout: float = 30.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 regular, bounded requests (in seconds).
            stream_timeout: Timeout for establishing a streaming connection (in
                seconds). Streaming endpoints are proxied through to a Docker
                host that may be slow to respond, so this is more generous than
                ``request_timeout``. Once established, the stream itself is not
                time-limited.
            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._stream_timeout = stream_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(_is_retryable),
            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 _open_stream(
        self,
        uri: str,
        *,
        method: str = METH_GET,
        params: dict[str, Any] | None = None,
    ) -> tuple[ClientResponse, URL]:
        """Establish a streaming connection and return the response and its URL.

        Args:
        ----
            uri: Request URI, without '/api/'.
            method: HTTP method to use.
            params: Query parameters to include in the request.

        Returns:
        -------
            A tuple of the open response and the URL it was requested from.

        The connection-establishment step is bounded by ``stream_timeout``. An
        explicit :class:`~aiohttp.ClientTimeout` without a ``total`` is passed to
        the request so the session's default total timeout (300 seconds for a
        plain :class:`~aiohttp.ClientSession`) cannot tear down a long-lived
        stream part-way through.

        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._stream_timeout):
                response = await self._session.request(
                    method,
                    url,
                    headers=headers,
                    params=params,
                    timeout=ClientTimeout(
                        total=None,
                        connect=self._stream_timeout,
                        sock_connect=self._stream_timeout,
                        sock_read=None,
                    ),
                )
                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

        return response, url

    async def _stream_request(
        self,
        uri: str,
        *,
        method: str = METH_GET,
        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 ``stream_timeout``;
        the ongoing stream is not time-limited.

        Args:
        ----
            uri: Request URI, without '/api/'.
            method: HTTP method to use.
            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.

        """
        response, url = await self._open_stream(uri, method=method, params=params)

        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 not stripped:
                        continue
                    try:
                        yield json.loads(stripped)
                    except json.JSONDecodeError as err:
                        _LOGGER.debug("Skipping malformed event line from %s: %s", url, err)
                        msg = f"Malformed JSON event from {url}: {err}"
                        raise PortainerError(msg) from err
        except TimeoutError as err:
            msg = f"Timeout while streaming from {url}: {err}"
            raise PortainerTimeoutError(msg) from err
        except (ClientError, socket.gaierror) as err:
            msg = f"Connection lost while streaming from {url}: {err}"
            raise PortainerConnectionError(msg) from err
        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.event.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.event.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.event.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 all containers, including stopped ones, from the Portainer API.

        Args:
        ----
            endpoint_id: The ID of the endpoint to get containers from.

        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),
            return_exceptions=True,
        )

        if isinstance(local, BaseException):
            raise local
        if isinstance(remote, PortainerConnectionError) and isinstance(remote.__cause__, ClientResponseError) and remote.__cause__.status == 403:
            _LOGGER.debug("No registry access for image %s on endpoint %s; skipping update check", image, endpoint_id)
            local_digest = next(
                (digest.partition("@")[2] for digest in (local.repo_digests or []) if "@" in digest),
                None,
            )
            return PortainerImageUpdateStatus(update_available=False, local_digest=local_digest, registry_digest=None)
        if isinstance(remote, BaseException):
            raise remote

        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",
            timeout=timeout.total_seconds(),
            method="POST",
            params=params,
            parse=False,
        )

    async def image_pull(self, endpoint_id: int, image: str) -> AsyncGenerator[DockerImagePullEvent, None]:
        """Pull a Docker image and stream its progress.

        Feed the events to :class:`~pyportainer.image_pull.ImagePullProgress`
        for an overall percentage. Portainer only adds stored registry
        credentials to pulls that name a registry, so private images may fail
        to pull here.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            image: The image to pull, for example ``nginx:latest``.

        Yields:
        ------
            :class:`~pyportainer.models.image_pull.DockerImagePullEvent` objects.

        Raises:
        ------
            PortainerImagePullError: If Docker reports an error during the pull.

        """
        async for raw in self._stream_request(
            f"endpoints/{endpoint_id}/docker/images/create",
            method=METH_POST,
            params={"fromImage": image},
        ):
            event = DockerImagePullEvent.from_dict(raw)
            if event.error:
                msg = f"Failed to pull image {image}: {event.error}"
                raise PortainerImagePullError(msg)
            yield event

    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 <timestamp> can be Unix timestamps, date formatted timestamps, or Go duration strings (e.g. 10m, 1h30m).

        Returns:
        -------
            The response from the Portainer API.

        """
        filters: dict[str, list[str]] = {"dangling": [str(dangling).lower()]}
        if until is not None:
            filters["until"] = [str(int((datetime.now(UTC) - until).timestamp()))]

        response = await self._request(
            f"endpoints/{endpoint_id}/docker/images/prune",
            method="POST",
            params={"filters": json.dumps(filters)},
        )

        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 portainer_system_version(self) -> PortainerSystemVersion:
        """Get the version of the Portainer instance and whether an update is available.

        Portainer looks up the latest release on GitHub on every call, so don't poll this often.

        Returns
        -------
            A PortainerSystemVersion object with the version data.

        """
        version = await self._request("system/version")

        return PortainerSystemVersion.from_dict(version)

    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 get_stack_file(self, stack_id: int) -> str:
        """Get the content of a stack's stack file.

        Args:
        ----
            stack_id: The ID of the stack.

        Returns:
        -------
            The stack file (for example a compose file) as text.

        """
        response = await self._request(f"stacks/{stack_id}/file")
        return str(response["StackFileContent"])

    async def update_stack(
        self,
        endpoint_id: int,
        stack_id: int,
        *,
        pull_image: bool = True,
        prune: bool | None = None,
        timeout: timedelta = timedelta(minutes=5),
    ) -> Stack:
        """Redeploy a stack, pulling the latest images first by default.

        Git-based stacks are redeployed from their repository. Other stacks are
        redeployed from their current stack file with their current environment
        variables, as Portainer replaces those with whatever an update sends.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            stack_id: The ID of the stack.
            pull_image: Whether to pull the latest images before redeploying.
            prune: Whether to remove services no longer in the stack file.
                Defaults to the stack's own prune setting.
            timeout: The timeout for redeploying the stack.

        Returns:
        -------
            Updated stack details.

        Raises:
        ------
            PortainerError: For Kubernetes stacks, which Portainer updates differently.

        """
        stack = await self.get_stack(stack_id)
        if stack.stack_type == StackType.KUBERNETES:
            msg = f"Updating Kubernetes stack {stack_id} is not supported"
            raise PortainerError(msg)

        # Portainer 2.36 renamed PullImage to RepullImageAndRedeploy; it still accepts both.
        json_body: dict[str, Any] = {
            "RepullImageAndRedeploy": pull_image,
            "PullImage": pull_image,
        }
        if stack.git_config is not None:
            uri = f"stacks/{stack_id}/git/redeploy"
            if prune is not None:
                json_body["Prune"] = prune
        else:
            uri = f"stacks/{stack_id}"
            json_body["StackFileContent"] = await self.get_stack_file(stack_id)
            json_body["Env"] = [{"name": env.name, "value": env.value} for env in stack.env or []]
            json_body["Prune"] = prune if prune is not None else bool(stack.option and stack.option.prune)

        response = await self._request(
            uri,
            method=METH_PUT,
            params={"endpointId": endpoint_id},
            json_body=json_body,
            timeout=timeout.total_seconds(),
        )
        return Stack.from_dict(response)

    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 also prune unused named volumes, not just anonymous ones.

        Returns:
        -------
            The response from the Portainer API.

        """
        # Docker API < 1.42 rejects the "all" filter, so only send it when needed.
        params = {"filters": json.dumps({"all": ["true"]})} if all_volumes else None
        return await self._request(f"endpoints/{endpoint_id}/docker/volumes/prune", method=METH_POST, params=params)

    async def prune_networks(self, endpoint_id: int, *, until: timedelta | None = None) -> DockerNetworkPruneResponse:
        """Prune unused networks on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            until: Only prune networks created longer ago than this.

        Returns:
        -------
            The response from the Portainer API.

        """
        params = None
        if until is not None:
            params = {"filters": json.dumps({"until": [str(int((datetime.now(UTC) - until).timestamp()))]})}

        response = await self._request(f"endpoints/{endpoint_id}/docker/networks/prune", method=METH_POST, params=params)

        return DockerNetworkPruneResponse.from_dict(response)

    async def prune_build_cache(self, endpoint_id: int, *, all_cache: bool = False, until: timedelta | None = None) -> DockerBuildCachePruneResponse:
        """Prune the Docker build cache on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            all_cache: Set to True to prune all unused build cache, not just dangling cache.
            until: Only prune build cache that was last used longer ago than this.

        Returns:
        -------
            The response from the Portainer API.

        """
        params: dict[str, str] = {}
        if all_cache:
            params["all"] = "true"
        if until is not None:
            params["filters"] = json.dumps({"until": [str(int((datetime.now(UTC) - until).timestamp()))]})

        response = await self._request(f"endpoints/{endpoint_id}/docker/build/prune", method=METH_POST, params=params or None)

        return DockerBuildCachePruneResponse.from_dict(response)

    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()
        # cgroup v2 hosts don't report percpu_usage, so prefer online_cpus like the Docker CLI.
        num_cpus = stats.cpu_stats.online_cpus or len(stats.cpu_stats.cpu_usage.percpu_usage) or 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 * scale
                if cpu_user_delta > 0:
                    docker_stats.cpu_user_percentage = cpu_user_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()

__aenter__() async

Async enter.

Returns

The Portainer object.

Source code in src/pyportainer/pyportainer.py

async def __aenter__(self) -> Self:
    """Async enter.

    Returns
    -------
        The Portainer object.

    """
    return self

__aexit__(*_exc_info) async

Async exit.

---

_exc_info: Exec type.

Source code in src/pyportainer/pyportainer.py

async def __aexit__(self, *_exc_info: object) -> None:
    """Async exit.

    Args:
    ----
        _exc_info: Exec type.

    """
    await self.close()

__init__(api_url, api_key, *, request_timeout=10.0, stream_timeout=30.0, session=None, max_retries=3)

Initialize the Portainer object.

---

api_url: URL of the Portainer API.
api_key: API key for authentication.
request_timeout: Timeout for regular, bounded requests (in seconds).
stream_timeout: Timeout for establishing a streaming connection (in
    seconds). Streaming endpoints are proxied through to a Docker
    host that may be slow to respond, so this is more generous than
    ``request_timeout``. Once established, the stream itself is not
    time-limited.
session: Optional aiohttp session to use.
max_retries: Maximum number of retry attempts on transient errors.

Source code in src/pyportainer/pyportainer.py

def __init__(  # pylint: disable=too-many-arguments
    self,
    api_url: str,
    api_key: str,
    *,
    request_timeout: float = 10.0,
    stream_timeout: float = 30.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 regular, bounded requests (in seconds).
        stream_timeout: Timeout for establishing a streaming connection (in
            seconds). Streaming endpoints are proxied through to a Docker
            host that may be slow to respond, so this is more generous than
            ``request_timeout``. Once established, the stream itself is not
            time-limited.
        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._stream_timeout = stream_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

close() async

Close open client session.

Source code in src/pyportainer/pyportainer.py

async def close(self) -> None:
    """Close open client session."""
    if self._session and self._close_session:
        await self._session.close()

container_create(endpoint_id, name, image, config) async

Create a Docker container.

---

endpoint_id: The ID of the endpoint.
name: The name of the container to create.

---

A DockerContainer object with the created container data.

Source code in src/pyportainer/pyportainer.py

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)

container_image_status(endpoint_id, image) async

Check whether a newer version of a Docker image is available in the registry.

---

endpoint_id: The ID of the endpoint.
image: The image name (with optional tag) to check.

---

A PortainerImageUpdateStatus with the comparison result and digests.

Source code in src/pyportainer/pyportainer.py

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),
        return_exceptions=True,
    )

    if isinstance(local, BaseException):
        raise local
    if isinstance(remote, PortainerConnectionError) and isinstance(remote.__cause__, ClientResponseError) and remote.__cause__.status == 403:
        _LOGGER.debug("No registry access for image %s on endpoint %s; skipping update check", image, endpoint_id)
        local_digest = next(
            (digest.partition("@")[2] for digest in (local.repo_digests or []) if "@" in digest),
            None,
        )
        return PortainerImageUpdateStatus(update_available=False, local_digest=local_digest, registry_digest=None)
    if isinstance(remote, BaseException):
        raise remote

    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,
    )

container_recreate(endpoint_id, container_id, timeout=timedelta(minutes=5), *, pull_image=False) async

Recreate a Docker container.

---

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.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

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)

container_recreate_helper(endpoint_id, container_id, image, timeout=timedelta(minutes=5)) async

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.

---

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.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

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

container_stats(endpoint_id, container_id, *, stream=False, one_shot=True) async

Get the stats of a container on the specified endpoint.

---

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.

---

The stats of the container.

Source code in src/pyportainer/pyportainer.py

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)

delete_container(endpoint_id, container_id, *, force=False) async

Delete a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to delete.
force: If True, force delete the container.

Source code in src/pyportainer/pyportainer.py

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,
    )

delete_stack(stack_id, endpoint_id, *, external=False) async

Delete a stack.

---

stack_id: The ID of the stack.
endpoint_id: The ID of the endpoint.
external: Set to True to delete an external Swarm stack.

Source code in src/pyportainer/pyportainer.py

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,
    )

docker_info(endpoint_id) async

Get the Docker info on the specified endpoint.

---

endpoint_id: The ID of the endpoint.

---

A DockerInfo object with the Docker info data.

Source code in src/pyportainer/pyportainer.py

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)

docker_system_df(endpoint_id, data_type=None, *, verbose=False) async

Get Docker system disk usage on the specified endpoint.

---

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.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

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)

docker_version(endpoint_id) async

Get the Docker version on the specified endpoint.

---

endpoint_id: The ID of the endpoint.

---

A DockerVersion object with the Docker version data.

Source code in src/pyportainer/pyportainer.py

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)

get_container_cpu_usage(endpoint_id, container_id) async

Get the current CPU usage percentage for the specified container.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container.

---

The current CPU usage as a percentage.

Source code in src/pyportainer/pyportainer.py

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()
    # cgroup v2 hosts don't report percpu_usage, so prefer online_cpus like the Docker CLI.
    num_cpus = stats.cpu_stats.online_cpus or len(stats.cpu_stats.cpu_usage.percpu_usage) or 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 * scale
            if cpu_user_delta > 0:
                docker_stats.cpu_user_percentage = cpu_user_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

get_containers(endpoint_id) async

Get all containers, including stopped ones, from the Portainer API.

---

endpoint_id: The ID of the endpoint to get containers from.

---

A list of containers.

Source code in src/pyportainer/pyportainer.py

async def get_containers(self, endpoint_id: int) -> list[DockerContainer]:
    """Get all containers, including stopped ones, from the Portainer API.

    Args:
    ----
        endpoint_id: The ID of the endpoint to get containers from.

    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]

get_endpoints() async

Get the list of endpoints from the Portainer API.

Returns

A list of Endpoint objects.

Source code in src/pyportainer/pyportainer.py

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]

get_events(endpoint_id, *, since=None, until=None, filters=None) async

Stream Docker events from an endpoint in real time.

Opens a persistent connection to the Docker events endpoint and yields :class:~pyportainer.models.event.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.

---

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"]}``.

---

:class:`~pyportainer.models.event.DockerEvent` objects.

Source code in src/pyportainer/pyportainer.py

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.event.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.event.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)

get_image(endpoint_id, image_id) async

Get information about a Docker image.

---

endpoint_id: The ID of the endpoint.
image_id: The ID of the image to get information about.

---

A LocalImageInformation object with the image data.

Source code in src/pyportainer/pyportainer.py

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)

get_image_information(endpoint_id, image_id) async

Get information about a Docker image.

---

endpoint_id: The ID of the endpoint.
image_id: The ID of the image to get information about.

---

An ImageInformation object with the image data.

Source code in src/pyportainer/pyportainer.py

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)

get_recent_events(endpoint_id, *, since, until=None, filters=None) async

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.

---

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"]}``.

---

A list of :class:`~pyportainer.models.event.DockerEvent` objects,
ordered as received from the Docker daemon.

Source code in src/pyportainer/pyportainer.py

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.event.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,
        )
    ]

get_stack(stack_id) async

Get details of a specific stack.

---

stack_id: The ID of the stack.

---

Stack details.

Source code in src/pyportainer/pyportainer.py

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)

get_stack_containers(endpoint_id, stack_name) async

Get containers belonging to a stack.

Filters containers by the com.docker.compose.project label.

---

endpoint_id: The ID of the endpoint.
stack_name: The name of the stack.

---

A list of containers in the stack.

Source code in src/pyportainer/pyportainer.py

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]

get_stack_file(stack_id) async

Get the content of a stack's stack file.

---

stack_id: The ID of the stack.

---

The stack file (for example a compose file) as text.

Source code in src/pyportainer/pyportainer.py

async def get_stack_file(self, stack_id: int) -> str:
    """Get the content of a stack's stack file.

    Args:
    ----
        stack_id: The ID of the stack.

    Returns:
    -------
        The stack file (for example a compose file) as text.

    """
    response = await self._request(f"stacks/{stack_id}/file")
    return str(response["StackFileContent"])

get_stacks(endpoint_id=None, swarm_id=None) async

Get the list of stacks from the Portainer API.

---

endpoint_id: Filter stacks by endpoint ID.
swarm_id: Filter stacks by Swarm cluster ID.

---

A list of stacks.

Source code in src/pyportainer/pyportainer.py

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]

get_volumes(endpoint_id) async

Get the list of volumes from the Portainer API.

---

endpoint_id: The ID of the endpoint to get volumes from.

---

A list of DockerVolume objects.

Source code in src/pyportainer/pyportainer.py

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 [])]

image_pull(endpoint_id, image) async

Pull a Docker image and stream its progress.

Feed the events to :class:~pyportainer.image_pull.ImagePullProgress for an overall percentage. Portainer only adds stored registry credentials to pulls that name a registry, so private images may fail to pull here.

---

endpoint_id: The ID of the endpoint.
image: The image to pull, for example ``nginx:latest``.

---

:class:`~pyportainer.models.image_pull.DockerImagePullEvent` objects.

---

PortainerImagePullError: If Docker reports an error during the pull.

Source code in src/pyportainer/pyportainer.py

async def image_pull(self, endpoint_id: int, image: str) -> AsyncGenerator[DockerImagePullEvent, None]:
    """Pull a Docker image and stream its progress.

    Feed the events to :class:`~pyportainer.image_pull.ImagePullProgress`
    for an overall percentage. Portainer only adds stored registry
    credentials to pulls that name a registry, so private images may fail
    to pull here.

    Args:
    ----
        endpoint_id: The ID of the endpoint.
        image: The image to pull, for example ``nginx:latest``.

    Yields:
    ------
        :class:`~pyportainer.models.image_pull.DockerImagePullEvent` objects.

    Raises:
    ------
        PortainerImagePullError: If Docker reports an error during the pull.

    """
    async for raw in self._stream_request(
        f"endpoints/{endpoint_id}/docker/images/create",
        method=METH_POST,
        params={"fromImage": image},
    ):
        event = DockerImagePullEvent.from_dict(raw)
        if event.error:
            msg = f"Failed to pull image {image}: {event.error}"
            raise PortainerImagePullError(msg)
        yield event

image_recreate(endpoint_id, image_id, timeout=timedelta(minutes=5)) async

Recreate a Docker image.

---

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.

---

An ImageInformation object with the recreated image data.

Source code in src/pyportainer/pyportainer.py

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",
        timeout=timeout.total_seconds(),
        method="POST",
        params=params,
        parse=False,
    )

images_prune(endpoint_id, until, *, dangling) async

Prune Docker images on the specified endpoint.

---

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 <timestamp> can be Unix timestamps, date formatted timestamps, or Go duration strings (e.g. 10m, 1h30m).

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

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 <timestamp> can be Unix timestamps, date formatted timestamps, or Go duration strings (e.g. 10m, 1h30m).

    Returns:
    -------
        The response from the Portainer API.

    """
    filters: dict[str, list[str]] = {"dangling": [str(dangling).lower()]}
    if until is not None:
        filters["until"] = [str(int((datetime.now(UTC) - until).timestamp()))]

    response = await self._request(
        f"endpoints/{endpoint_id}/docker/images/prune",
        method="POST",
        params={"filters": json.dumps(filters)},
    )

    return DockerImagePruneResponse.from_dict(response)

inspect_container(endpoint_id, container_id, *, raw=False) async

Inspect a container on the specified endpoint.

---

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.

---

A DockerContainer object with the inspected data.

Source code in src/pyportainer/pyportainer.py

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)

inspect_volume(endpoint_id, volume_name) async

Inspect a volume on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
volume_name: The name of the volume to inspect.

---

A DockerVolume object with the inspected volume data.

Source code in src/pyportainer/pyportainer.py

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)

kill_container(endpoint_id, container_id) async

Kill a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to kill.

Source code in src/pyportainer/pyportainer.py

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",
    )

pause_container(endpoint_id, container_id) async

Pause a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to pause.

Source code in src/pyportainer/pyportainer.py

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",
    )

portainer_system_status() async

Get the system status of the Portainer instance.

Returns

A PortainerSystemStatus object with the system status data.

Source code in src/pyportainer/pyportainer.py

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)

portainer_system_version() async

Get the version of the Portainer instance and whether an update is available.

Portainer looks up the latest release on GitHub on every call, so don't poll this often.

Returns

A PortainerSystemVersion object with the version data.

Source code in src/pyportainer/pyportainer.py

async def portainer_system_version(self) -> PortainerSystemVersion:
    """Get the version of the Portainer instance and whether an update is available.

    Portainer looks up the latest release on GitHub on every call, so don't poll this often.

    Returns
    -------
        A PortainerSystemVersion object with the version data.

    """
    version = await self._request("system/version")

    return PortainerSystemVersion.from_dict(version)

prune_build_cache(endpoint_id, *, all_cache=False, until=None) async

Prune the Docker build cache on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
all_cache: Set to True to prune all unused build cache, not just dangling cache.
until: Only prune build cache that was last used longer ago than this.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

async def prune_build_cache(self, endpoint_id: int, *, all_cache: bool = False, until: timedelta | None = None) -> DockerBuildCachePruneResponse:
    """Prune the Docker build cache on the specified endpoint.

    Args:
    ----
        endpoint_id: The ID of the endpoint.
        all_cache: Set to True to prune all unused build cache, not just dangling cache.
        until: Only prune build cache that was last used longer ago than this.

    Returns:
    -------
        The response from the Portainer API.

    """
    params: dict[str, str] = {}
    if all_cache:
        params["all"] = "true"
    if until is not None:
        params["filters"] = json.dumps({"until": [str(int((datetime.now(UTC) - until).timestamp()))]})

    response = await self._request(f"endpoints/{endpoint_id}/docker/build/prune", method=METH_POST, params=params or None)

    return DockerBuildCachePruneResponse.from_dict(response)

prune_networks(endpoint_id, *, until=None) async

Prune unused networks on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
until: Only prune networks created longer ago than this.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

async def prune_networks(self, endpoint_id: int, *, until: timedelta | None = None) -> DockerNetworkPruneResponse:
    """Prune unused networks on the specified endpoint.

    Args:
    ----
        endpoint_id: The ID of the endpoint.
        until: Only prune networks created longer ago than this.

    Returns:
    -------
        The response from the Portainer API.

    """
    params = None
    if until is not None:
        params = {"filters": json.dumps({"until": [str(int((datetime.now(UTC) - until).timestamp()))]})}

    response = await self._request(f"endpoints/{endpoint_id}/docker/networks/prune", method=METH_POST, params=params)

    return DockerNetworkPruneResponse.from_dict(response)

prune_volumes(endpoint_id, *, all_volumes=False) async

Prune unused volumes on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
all_volumes: Set to True to also prune unused named volumes, not just anonymous ones.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

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 also prune unused named volumes, not just anonymous ones.

    Returns:
    -------
        The response from the Portainer API.

    """
    # Docker API < 1.42 rejects the "all" filter, so only send it when needed.
    params = {"filters": json.dumps({"all": ["true"]})} if all_volumes else None
    return await self._request(f"endpoints/{endpoint_id}/docker/volumes/prune", method=METH_POST, params=params)

restart_container(endpoint_id, container_id) async

Restart a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to restart.

Source code in src/pyportainer/pyportainer.py

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",
    )

start_container(endpoint_id, container_id) async

Start a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to start.

Source code in src/pyportainer/pyportainer.py

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={},
    )

start_stack(endpoint_id, stack_id, timeout=timedelta(minutes=5)) async

Start a stopped stack.

---

stack_id: The ID of the stack.
endpoint_id: The ID of the endpoint.
timeout: The timeout for starting the stack.

---

Updated stack details.

Source code in src/pyportainer/pyportainer.py

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)

stop_container(endpoint_id, container_id) async

Stop a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to stop.

Source code in src/pyportainer/pyportainer.py

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",
    )

stop_stack(endpoint_id, stack_id, timeout=timedelta(minutes=5)) async

Stop a running stack.

---

stack_id: The ID of the stack.
endpoint_id: The ID of the endpoint.
timeout: The timeout for stopping the stack.

---

Updated stack details.

Source code in src/pyportainer/pyportainer.py

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)

unpause_container(endpoint_id, container_id) async

Unpause a container on the specified endpoint.

---

endpoint_id: The ID of the endpoint.
container_id: The ID of the container to unpause.

Source code in src/pyportainer/pyportainer.py

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",
    )

update_stack(endpoint_id, stack_id, *, pull_image=True, prune=None, timeout=timedelta(minutes=5)) async

Redeploy a stack, pulling the latest images first by default.

Git-based stacks are redeployed from their repository. Other stacks are redeployed from their current stack file with their current environment variables, as Portainer replaces those with whatever an update sends.

---

endpoint_id: The ID of the endpoint.
stack_id: The ID of the stack.
pull_image: Whether to pull the latest images before redeploying.
prune: Whether to remove services no longer in the stack file.
    Defaults to the stack's own prune setting.
timeout: The timeout for redeploying the stack.

---

Updated stack details.

---

PortainerError: For Kubernetes stacks, which Portainer updates differently.

Source code in src/pyportainer/pyportainer.py

async def update_stack(
    self,
    endpoint_id: int,
    stack_id: int,
    *,
    pull_image: bool = True,
    prune: bool | None = None,
    timeout: timedelta = timedelta(minutes=5),
) -> Stack:
    """Redeploy a stack, pulling the latest images first by default.

    Git-based stacks are redeployed from their repository. Other stacks are
    redeployed from their current stack file with their current environment
    variables, as Portainer replaces those with whatever an update sends.

    Args:
    ----
        endpoint_id: The ID of the endpoint.
        stack_id: The ID of the stack.
        pull_image: Whether to pull the latest images before redeploying.
        prune: Whether to remove services no longer in the stack file.
            Defaults to the stack's own prune setting.
        timeout: The timeout for redeploying the stack.

    Returns:
    -------
        Updated stack details.

    Raises:
    ------
        PortainerError: For Kubernetes stacks, which Portainer updates differently.

    """
    stack = await self.get_stack(stack_id)
    if stack.stack_type == StackType.KUBERNETES:
        msg = f"Updating Kubernetes stack {stack_id} is not supported"
        raise PortainerError(msg)

    # Portainer 2.36 renamed PullImage to RepullImageAndRedeploy; it still accepts both.
    json_body: dict[str, Any] = {
        "RepullImageAndRedeploy": pull_image,
        "PullImage": pull_image,
    }
    if stack.git_config is not None:
        uri = f"stacks/{stack_id}/git/redeploy"
        if prune is not None:
            json_body["Prune"] = prune
    else:
        uri = f"stacks/{stack_id}"
        json_body["StackFileContent"] = await self.get_stack_file(stack_id)
        json_body["Env"] = [{"name": env.name, "value": env.value} for env in stack.env or []]
        json_body["Prune"] = prune if prune is not None else bool(stack.option and stack.option.prune)

    response = await self._request(
        uri,
        method=METH_PUT,
        params={"endpointId": endpoint_id},
        json_body=json_body,
        timeout=timeout.total_seconds(),
    )
    return Stack.from_dict(response)

PortainerAuthenticationError

Bases: PortainerError

Exception raised for authentication errors.

Source code in src/pyportainer/exceptions.py

class PortainerAuthenticationError(PortainerError):
    """Exception raised for authentication errors."""

PortainerConnectionError

Bases: PortainerError

Exception raised for connection errors.

Source code in src/pyportainer/exceptions.py

class PortainerConnectionError(PortainerError):
    """Exception raised for connection errors."""

PortainerError

Bases: Exception

Generic exception for Portainer errors.

Source code in src/pyportainer/exceptions.py

class PortainerError(Exception):
    """Generic exception for Portainer errors."""

PortainerEventListener

Maintains persistent streaming connections to Docker event endpoints.

One streaming connection is opened per endpoint. Events are delivered to registered callbacks as they arrive, in real time. If a connection drops, it is automatically re-established after reconnect_interval, with the delay doubling on each consecutive failure up to max_reconnect_interval.

Source code in src/pyportainer/listener.py

class PortainerEventListener:
    """Maintains persistent streaming connections to Docker event endpoints.

    One streaming connection is opened per endpoint. Events are delivered to
    registered callbacks as they arrive, in real time. If a connection drops,
    it is automatically re-established after ``reconnect_interval``, with the
    delay doubling on each consecutive failure up to ``max_reconnect_interval``.
    """

    def __init__(  # pylint: disable=too-many-arguments,too-many-instance-attributes
        self,
        portainer: Portainer,
        endpoint_id: int | None = None,
        *,
        event_types: list[str] | None = None,
        reconnect_interval: timedelta = timedelta(seconds=5),
        max_reconnect_interval: timedelta = timedelta(minutes=5),
        debug: bool = False,
    ) -> None:
        """Initialize the PortainerEventListener.

        Args:
        ----
            portainer: An authenticated Portainer client instance.
            endpoint_id: The ID of the endpoint to listen to. If None, all
                endpoints are monitored concurrently.
            event_types: Docker event types to filter on, e.g.
                ``["container", "image"]``. If None, all event types are
                delivered.
            reconnect_interval: How long to wait before the first reconnect
                attempt after a dropped connection. Defaults to 5 seconds. The
                delay doubles on each consecutive failure.
            max_reconnect_interval: Upper bound on the reconnect delay.
                Defaults to 5 minutes.
            debug: Raise this logger's level to DEBUG. Logging is otherwise
                left entirely to the application; the level configured by the
                caller is never lowered or overwritten.

        """
        self._portainer = portainer
        self._endpoint_id = endpoint_id
        self._event_types = event_types
        self._reconnect_interval = reconnect_interval
        self._max_reconnect_interval = max_reconnect_interval
        self._task: asyncio.Task[None] | None = None
        self._callbacks: list[EventListenerCallback] = []
        self._failures: dict[int | str, int] = {}

        if debug:
            _LOGGER.setLevel(logging.DEBUG)

    def start(self) -> None:
        """Start listening for Docker events.

        Opens streaming connections immediately. Must be called from within a
        running asyncio event loop.
        """
        if self._task is None or self._task.done():
            self._task = asyncio.create_task(self._run())

    def stop(self) -> None:
        """Stop all streaming connections."""
        if self._task and not self._task.done():
            self._task.cancel()

    def register_callback(self, callback: EventListenerCallback) -> None:
        """Register a callback to be invoked for every Docker event received.

        Both synchronous and async callables are supported. The callback
        receives a single :class:`PortainerEventListenerResult` argument.
        Each unique callable is only registered once; duplicates are ignored.

        Args:
        ----
            callback: A sync or async callable that accepts a
                :class:`PortainerEventListenerResult`.

        """
        if callback not in self._callbacks:
            self._callbacks.append(callback)

    def unregister_callback(self, callback: EventListenerCallback) -> None:
        """Remove a previously registered callback.

        Args:
        ----
            callback: The callable to remove. Raises :exc:`ValueError` if it
                was not registered.

        """
        self._callbacks.remove(callback)

    async def _fire_callbacks(self, result: PortainerEventListenerResult) -> None:
        """Invoke all registered callbacks for a single event.

        Exceptions raised by individual callbacks are logged but do not stop
        the listener.
        """
        for callback in list(self._callbacks):
            try:
                ret = callback(result)
                if asyncio.iscoroutine(ret):
                    await ret
            except Exception:  # pylint: disable=broad-except
                _LOGGER.exception(
                    "Callback raised an exception for event %s on endpoint %s",
                    result.event.action,
                    result.endpoint_id,
                )

    async def _backoff(self, key: int | str, message: str, err: Exception, *, exc_info: bool = False) -> None:
        """Log a failed attempt and sleep out the reconnect delay.

        The delay doubles with each consecutive failure on ``key``, capped at
        ``max_reconnect_interval``, with a little jitter so several endpoints
        failing at once don't all retry in lockstep. Only the first failure of
        a streak is logged at warning level; the rest are logged at debug
        level so a persistently unreachable endpoint doesn't flood the log.

        Args:
        ----
            key: Endpoint ID, or :data:`_DISCOVERY_KEY` for endpoint discovery.
            message: What went wrong, e.g. ``"Timeout on endpoint 5"``.
            err: The exception that ended the attempt.
            exc_info: Whether to log a traceback with the first warning.

        """
        failures = self._failures.get(key, 0) + 1
        self._failures[key] = failures

        base = self._reconnect_interval.total_seconds()
        ceiling = max(base, self._max_reconnect_interval.total_seconds())
        delay = min(base * 2 ** (failures - 1), ceiling) * random.uniform(0.8, 1.0)  # noqa: S311 - jitter, not security

        verb = "retrying" if key == _DISCOVERY_KEY else "reconnecting"
        if failures == 1:
            _LOGGER.warning("%s, %s in %.1fs: %s", message, verb, delay, err, exc_info=exc_info)
        else:
            _LOGGER.debug("%s, %s in %.1fs (attempt %d): %s", message, verb, delay, failures, err)

        await asyncio.sleep(delay)

    def _reset_failures(self, key: int | str) -> None:
        """Clear a failure streak, e.g. after a successful reconnect."""
        if self._failures.pop(key, None):
            _LOGGER.info("Recovered on %s after previous failures", "endpoint discovery" if key == _DISCOVERY_KEY else f"endpoint {key}")

    async def _listen(self, endpoint_id: int) -> None:
        """Stream events from a single endpoint and fire callbacks for each.

        Args:
        ----
            endpoint_id: The endpoint to stream events from.

        """
        filters = {"type": self._event_types} if self._event_types else None
        async for event in self._portainer.get_events(endpoint_id, filters=filters):
            self._reset_failures(endpoint_id)
            result = PortainerEventListenerResult(endpoint_id=endpoint_id, event=event)
            await self._fire_callbacks(result)

    async def _listen_with_reconnect(self, endpoint_id: int) -> None:
        """Stream events from an endpoint, reconnecting on transient errors.

        Authentication errors are treated as fatal and stop the listener for
        that endpoint. All other :class:`~pyportainer.exceptions.PortainerError`
        subclasses trigger a reconnect after a backoff delay; see
        :meth:`_backoff` for the backoff and log-throttling policy.

        Args:
        ----
            endpoint_id: The endpoint to stream events from.

        """
        while True:
            try:
                await self._listen(endpoint_id)
            except PortainerAuthenticationError:
                _LOGGER.exception(
                    "Authentication error for endpoint %s, stopping listener",
                    endpoint_id,
                )
                return
            except PortainerTimeoutError as err:
                await self._backoff(endpoint_id, f"Timeout on endpoint {endpoint_id}", err)
            except PortainerConnectionError as err:
                await self._backoff(endpoint_id, f"Connection lost on endpoint {endpoint_id}", err)
            except PortainerError as err:
                await self._backoff(endpoint_id, f"Error on endpoint {endpoint_id}", err, exc_info=True)

    async def _resolve_endpoint_ids(self) -> list[int]:
        """Resolve the list of endpoint IDs to listen to.

        Returns
        -------
            The list of endpoint IDs to listen to.

        """
        if self._endpoint_id is not None:
            return [self._endpoint_id]

        _LOGGER.debug("No endpoint_id specified, fetching all endpoints to listen to.")
        while True:
            try:
                endpoints = await self._portainer.get_endpoints()
            except PortainerTimeoutError as err:
                await self._backoff(_DISCOVERY_KEY, "Timeout fetching endpoints", err)
            except PortainerConnectionError as err:
                await self._backoff(_DISCOVERY_KEY, "Connection error fetching endpoints", err)
            else:
                self._reset_failures(_DISCOVERY_KEY)
                return [endpoint.id for endpoint in endpoints]

    async def _run(self) -> None:
        """Resolve endpoints and open a streaming connection to each.

        Endpoint discovery (when no explicit ``endpoint_id`` was supplied)
        retries indefinitely on transient errors; see
        :meth:`_resolve_endpoint_ids`.

        """
        endpoint_ids = await self._resolve_endpoint_ids()

        results = await asyncio.gather(
            *(self._listen_with_reconnect(ep_id) for ep_id in endpoint_ids),
            return_exceptions=True,
        )

        for endpoint_id, result in zip(endpoint_ids, results, strict=True):
            if isinstance(result, BaseException):
                _LOGGER.error(
                    "Listener for endpoint %s terminated unexpectedly",
                    endpoint_id,
                    exc_info=result,
                )

__init__(portainer, endpoint_id=None, *, event_types=None, reconnect_interval=timedelta(seconds=5), max_reconnect_interval=timedelta(minutes=5), debug=False)

Initialize the PortainerEventListener.

---

portainer: An authenticated Portainer client instance.
endpoint_id: The ID of the endpoint to listen to. If None, all
    endpoints are monitored concurrently.
event_types: Docker event types to filter on, e.g.
    ``["container", "image"]``. If None, all event types are
    delivered.
reconnect_interval: How long to wait before the first reconnect
    attempt after a dropped connection. Defaults to 5 seconds. The
    delay doubles on each consecutive failure.
max_reconnect_interval: Upper bound on the reconnect delay.
    Defaults to 5 minutes.
debug: Raise this logger's level to DEBUG. Logging is otherwise
    left entirely to the application; the level configured by the
    caller is never lowered or overwritten.

Source code in src/pyportainer/listener.py

def __init__(  # pylint: disable=too-many-arguments,too-many-instance-attributes
    self,
    portainer: Portainer,
    endpoint_id: int | None = None,
    *,
    event_types: list[str] | None = None,
    reconnect_interval: timedelta = timedelta(seconds=5),
    max_reconnect_interval: timedelta = timedelta(minutes=5),
    debug: bool = False,
) -> None:
    """Initialize the PortainerEventListener.

    Args:
    ----
        portainer: An authenticated Portainer client instance.
        endpoint_id: The ID of the endpoint to listen to. If None, all
            endpoints are monitored concurrently.
        event_types: Docker event types to filter on, e.g.
            ``["container", "image"]``. If None, all event types are
            delivered.
        reconnect_interval: How long to wait before the first reconnect
            attempt after a dropped connection. Defaults to 5 seconds. The
            delay doubles on each consecutive failure.
        max_reconnect_interval: Upper bound on the reconnect delay.
            Defaults to 5 minutes.
        debug: Raise this logger's level to DEBUG. Logging is otherwise
            left entirely to the application; the level configured by the
            caller is never lowered or overwritten.

    """
    self._portainer = portainer
    self._endpoint_id = endpoint_id
    self._event_types = event_types
    self._reconnect_interval = reconnect_interval
    self._max_reconnect_interval = max_reconnect_interval
    self._task: asyncio.Task[None] | None = None
    self._callbacks: list[EventListenerCallback] = []
    self._failures: dict[int | str, int] = {}

    if debug:
        _LOGGER.setLevel(logging.DEBUG)

register_callback(callback)

Register a callback to be invoked for every Docker event received.

Both synchronous and async callables are supported. The callback receives a single :class:PortainerEventListenerResult argument. Each unique callable is only registered once; duplicates are ignored.

---

callback: A sync or async callable that accepts a
    :class:`PortainerEventListenerResult`.

Source code in src/pyportainer/listener.py

def register_callback(self, callback: EventListenerCallback) -> None:
    """Register a callback to be invoked for every Docker event received.

    Both synchronous and async callables are supported. The callback
    receives a single :class:`PortainerEventListenerResult` argument.
    Each unique callable is only registered once; duplicates are ignored.

    Args:
    ----
        callback: A sync or async callable that accepts a
            :class:`PortainerEventListenerResult`.

    """
    if callback not in self._callbacks:
        self._callbacks.append(callback)

start()

Start listening for Docker events.

Opens streaming connections immediately. Must be called from within a running asyncio event loop.

Source code in src/pyportainer/listener.py

def start(self) -> None:
    """Start listening for Docker events.

    Opens streaming connections immediately. Must be called from within a
    running asyncio event loop.
    """
    if self._task is None or self._task.done():
        self._task = asyncio.create_task(self._run())

stop()

Stop all streaming connections.

Source code in src/pyportainer/listener.py

def stop(self) -> None:
    """Stop all streaming connections."""
    if self._task and not self._task.done():
        self._task.cancel()

unregister_callback(callback)

Remove a previously registered callback.

---

callback: The callable to remove. Raises :exc:`ValueError` if it
    was not registered.

Source code in src/pyportainer/listener.py

def unregister_callback(self, callback: EventListenerCallback) -> None:
    """Remove a previously registered callback.

    Args:
    ----
        callback: The callable to remove. Raises :exc:`ValueError` if it
            was not registered.

    """
    self._callbacks.remove(callback)

PortainerEventListenerResult dataclass

Represents a single Docker event received from an endpoint.

Source code in src/pyportainer/listener.py

@dataclass(frozen=True)
class PortainerEventListenerResult:
    """Represents a single Docker event received from an endpoint."""

    endpoint_id: int
    event: DockerEvent

PortainerImagePullError

Bases: PortainerError

Exception raised when Docker reports an error while pulling an image.

Source code in src/pyportainer/exceptions.py

class PortainerImagePullError(PortainerError):
    """Exception raised when Docker reports an error while pulling an image."""

PortainerImageWatcher

Periodically checks all containers on an endpoint for image updates.

Results are stored and accessible via the results property after each check.

Source code in src/pyportainer/watcher.py

class PortainerImageWatcher:
    """Periodically checks all containers on an endpoint for image updates.

    Results are stored and accessible via the `results` property after each check.
    """

    def __init__(
        self,
        portainer: Portainer,
        endpoint_id: int | None = None,
        interval: timedelta = timedelta(hours=12),
        *,
        debug: bool = False,
    ) -> None:
        """Initialize the PortainerImageWatcher.

        Args:
        ----
            portainer: An authenticated Portainer client instance.
            endpoint_id: The ID of the endpoint whose containers to monitor. If None, all endpoints are monitored.
            interval: How often to poll for updates. Defaults to 12 hours.
            debug: Raise this logger's level to DEBUG. Logging is otherwise left
                entirely to the application; the level configured by the caller
                is never lowered or overwritten.

        """
        self._portainer = portainer
        self._endpoint_id = endpoint_id
        self._interval = interval
        self._results: dict[tuple[int, str], PortainerImageWatcherResult] = {}
        self._task: asyncio.Task[None] | None = None
        self._last_check: float | None = None
        self._callbacks: list[WatcherCallback] = []

        if debug:
            _LOGGER.setLevel(logging.DEBUG)

    @property
    def interval(self) -> timedelta:
        """Polling interval."""
        return self._interval

    @interval.setter
    def interval(self, value: timedelta) -> None:
        """Update the polling interval. Takes effect after the next check."""
        self._interval = value

    @property
    def last_check(self) -> float | None:
        """Timestamp of the last completed check, or None if no checks have completed yet."""
        return self._last_check

    @property
    def results(self) -> dict[tuple[int, str], PortainerImageWatcherResult]:
        """Latest update status as of the last check."""
        return self._results.copy()

    def start(self) -> None:
        """Start the background polling loop.

        The first check runs immediately; subsequent checks run after each interval.
        Must be called from within a running asyncio event loop.
        """
        if self._task is None or self._task.done():
            self._task = asyncio.create_task(self._run())

    def stop(self) -> None:
        """Cancel the background polling loop."""
        if self._task and not self._task.done():
            self._task.cancel()

    def register_callback(self, callback: WatcherCallback) -> None:
        """Register a callback to be invoked for every result after each poll cycle.

        Both synchronous and async callables are supported. The callback receives a
        single :class:`PortainerImageWatcherResult` argument. Each unique callable
        is only registered once; duplicate registrations are silently ignored.

        Args:
        ----
            callback: A sync or async callable that accepts a
                :class:`PortainerImageWatcherResult`.

        """
        if callback not in self._callbacks:
            self._callbacks.append(callback)

    def unregister_callback(self, callback: WatcherCallback) -> None:
        """Remove a previously registered callback.

        Args:
        ----
            callback: The callable to remove. Raises :exc:`ValueError` if it was not registered.

        """
        self._callbacks.remove(callback)

    async def _fire_callbacks(self, result: PortainerImageWatcherResult) -> None:
        """Invoke all registered callbacks for a single result.

        Exceptions raised by individual callbacks are logged but not blocking.
        """
        for callback in list(self._callbacks):
            try:
                ret = callback(result)
                if asyncio.iscoroutine(ret):
                    await ret
            except Exception:  # pylint: disable=broad-except
                _LOGGER.exception("Callback raised an exception for container %s", result.container_id)

    async def _run(self) -> None:
        """Loop that checks immediately, then sleep for the interval, then repeat.

        Errors during checks are logged but don't stop the watcher, allowing recovery from transient issues.
        """
        while True:
            try:
                await self._check_all()
            except PortainerTimeoutError:
                _LOGGER.exception("Timeout during image check")
            except PortainerConnectionError:
                _LOGGER.exception("Connection error during image check")
            except PortainerAuthenticationError:
                _LOGGER.exception("Authentication error during image check")
            except PortainerError:
                _LOGGER.exception("Error during image check")
            finally:
                self._last_check = time.time()

            await asyncio.sleep(self._interval.total_seconds())

    async def _check_all(self) -> None:
        """Fetch all containers and check each unique image concurrently.

        Errors for individual images are logged but silently skipped so one
        failing image does not prevent the rest from being checked.
        """
        if self._endpoint_id is not None:
            endpoint_ids: list[int] = [self._endpoint_id]
        else:
            _LOGGER.debug("No endpoint_id specified, fetching all endpoints to check.")
            endpoints = await self._portainer.get_endpoints()
            endpoint_ids = [endpoint.id for endpoint in endpoints]

        fresh: dict[tuple[int, str], PortainerImageWatcherResult] = {}

        for endpoint_id in endpoint_ids:
            try:
                containers = await self._portainer.get_containers(endpoint_id)
            except PortainerError:
                _LOGGER.warning("Failed to fetch containers for endpoint %s, skipping", endpoint_id)
                continue

            image_containers = defaultdict(list)
            for container in containers:
                if container.image and container.state == "running":
                    image_containers[container.image].append(container.id)

            _LOGGER.debug("Checking %d unique images for endpoint %s...", len(image_containers), endpoint_id)

            statuses = await asyncio.gather(
                *(self._portainer.container_image_status(endpoint_id, image) for image in image_containers),
                return_exceptions=True,
            )
            for image, status in zip(image_containers, statuses, strict=False):
                if isinstance(status, BaseException):
                    _LOGGER.warning("Failed to check image %s on endpoint %s: %s", image, endpoint_id, status)
                    continue
                for container_id in image_containers[image]:
                    fresh[(endpoint_id, container_id)] = PortainerImageWatcherResult(
                        endpoint_id=endpoint_id,
                        container_id=container_id,
                        status=status,
                    )

                    _LOGGER.debug("Checked image %s on endpoint %s for container %s", image, endpoint_id, container_id)

        self._results = fresh

        if self._callbacks and fresh:
            await asyncio.gather(*(self._fire_callbacks(result) for result in fresh.values()))

interval property writable

Polling interval.

last_check property

Timestamp of the last completed check, or None if no checks have completed yet.

results property

Latest update status as of the last check.

__init__(portainer, endpoint_id=None, interval=timedelta(hours=12), *, debug=False)

Initialize the PortainerImageWatcher.

---

portainer: An authenticated Portainer client instance.
endpoint_id: The ID of the endpoint whose containers to monitor. If None, all endpoints are monitored.
interval: How often to poll for updates. Defaults to 12 hours.
debug: Raise this logger's level to DEBUG. Logging is otherwise left
    entirely to the application; the level configured by the caller
    is never lowered or overwritten.

Source code in src/pyportainer/watcher.py

def __init__(
    self,
    portainer: Portainer,
    endpoint_id: int | None = None,
    interval: timedelta = timedelta(hours=12),
    *,
    debug: bool = False,
) -> None:
    """Initialize the PortainerImageWatcher.

    Args:
    ----
        portainer: An authenticated Portainer client instance.
        endpoint_id: The ID of the endpoint whose containers to monitor. If None, all endpoints are monitored.
        interval: How often to poll for updates. Defaults to 12 hours.
        debug: Raise this logger's level to DEBUG. Logging is otherwise left
            entirely to the application; the level configured by the caller
            is never lowered or overwritten.

    """
    self._portainer = portainer
    self._endpoint_id = endpoint_id
    self._interval = interval
    self._results: dict[tuple[int, str], PortainerImageWatcherResult] = {}
    self._task: asyncio.Task[None] | None = None
    self._last_check: float | None = None
    self._callbacks: list[WatcherCallback] = []

    if debug:
        _LOGGER.setLevel(logging.DEBUG)

register_callback(callback)

Register a callback to be invoked for every result after each poll cycle.

Both synchronous and async callables are supported. The callback receives a single :class:PortainerImageWatcherResult argument. Each unique callable is only registered once; duplicate registrations are silently ignored.

---

callback: A sync or async callable that accepts a
    :class:`PortainerImageWatcherResult`.

Source code in src/pyportainer/watcher.py

def register_callback(self, callback: WatcherCallback) -> None:
    """Register a callback to be invoked for every result after each poll cycle.

    Both synchronous and async callables are supported. The callback receives a
    single :class:`PortainerImageWatcherResult` argument. Each unique callable
    is only registered once; duplicate registrations are silently ignored.

    Args:
    ----
        callback: A sync or async callable that accepts a
            :class:`PortainerImageWatcherResult`.

    """
    if callback not in self._callbacks:
        self._callbacks.append(callback)

start()

Start the background polling loop.

The first check runs immediately; subsequent checks run after each interval. Must be called from within a running asyncio event loop.

Source code in src/pyportainer/watcher.py

def start(self) -> None:
    """Start the background polling loop.

    The first check runs immediately; subsequent checks run after each interval.
    Must be called from within a running asyncio event loop.
    """
    if self._task is None or self._task.done():
        self._task = asyncio.create_task(self._run())

stop()

Cancel the background polling loop.

Source code in src/pyportainer/watcher.py

def stop(self) -> None:
    """Cancel the background polling loop."""
    if self._task and not self._task.done():
        self._task.cancel()

unregister_callback(callback)

Remove a previously registered callback.

---

callback: The callable to remove. Raises :exc:`ValueError` if it was not registered.

Source code in src/pyportainer/watcher.py

def unregister_callback(self, callback: WatcherCallback) -> None:
    """Remove a previously registered callback.

    Args:
    ----
        callback: The callable to remove. Raises :exc:`ValueError` if it was not registered.

    """
    self._callbacks.remove(callback)

PortainerTimeoutError

Bases: PortainerError

Exception raised for timeout errors.

Source code in src/pyportainer/exceptions.py

class PortainerTimeoutError(PortainerError):
    """Exception raised for timeout errors."""

StackStatus

Bases: IntEnum

Portainer stack status.

Source code in src/pyportainer/models/docker.py

class StackStatus(IntEnum):
    """Portainer stack status."""

    ACTIVE = 1
    INACTIVE = 2

StackType

Bases: IntEnum

Portainer stack type.

Source code in src/pyportainer/models/docker.py

class StackType(IntEnum):
    """Portainer stack type."""

    SWARM = 1
    COMPOSE = 2
    KUBERNETES = 3