API Reference

Asynchronous Python client for Python Portainer.

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
    session: ClientSession | None = None

    _close_session: bool = False

    def __init__(  # pylint: disable=too-many-arguments
        self,
        api_url: str,
        api_key: str,
        *,
        request_timeout: float = 10.0,
        session: ClientSession | None = None,
        max_retries: int = 3,
    ) -> None:
        """Initialize the Portainer object.

        Args:
        ----
            api_url: URL of the Portainer API.
            api_key: API key for authentication.
            request_timeout: Timeout for requests (in seconds).
            session: Optional aiohttp session to use.
            max_retries: Maximum number of retry attempts on transient errors.

        """
        self._api_key = api_key
        self._request_timeout = request_timeout
        self._session = session
        self._max_retries = max_retries

        parsed_url = urlparse(api_url)
        self._api_host = parsed_url.hostname or ""
        self._api_scheme = parsed_url.scheme or ""
        self._api_port = parsed_url.port

        self._api_base_path = (parsed_url.path or "").rstrip("/")

    # pylint: disable=too-many-arguments, too-many-locals, too-many-branches
    async def _request(
        self,
        uri: str,
        *,
        method: str = METH_GET,
        params: dict[str, Any] | None = None,
        json_body: dict[str, Any] | None = None,
        timeout: float | None = None,
        parse: bool = True,
    ) -> Any:
        """Handle a request to the Python Portainer API.

        Args:
        ----
            uri: Request URI, without '/api/', for example, 'status'.
            method: HTTP method to use.
            params: Extra options to improve or limit the response.
            timeout: Timeout for the request (in seconds).
            parse: Whether to parse the response as JSON.

        Returns:
        -------
            A Python dictionary (JSON decoded) with the response from
            the Python Portainer API.

        Raises:
        ------
            Python PortainerAuthenticationError: If the API key is invalid.

        """
        url = URL.build(
            scheme=self._api_scheme,
            host=self._api_host,
            port=self._api_port,
            path=f"{self._api_base_path}/api/",
        ).join(URL(uri))

        headers = {
            "Accept": "application/json, text/plain",
            "User-Agent": f"PythonPortainer/{VERSION}",
            "X-API-Key": self._api_key,
        }

        if self._session is None:
            self._session = ClientSession()
            self._close_session = True

        # Only override timeout if a specific value is provided, else use default
        if timeout is None:
            timeout = self._request_timeout

        async for attempt in AsyncRetrying(
            retry=retry_if_exception_type((PortainerConnectionError, PortainerTimeoutError)),
            wait=wait_exponential(multiplier=1, min=1, max=10),
            stop=stop_after_attempt(self._max_retries + 1),
            reraise=True,
        ):
            with attempt:
                try:
                    async with asyncio.timeout(timeout):
                        response = await self._session.request(
                            method,
                            url,
                            headers=headers,
                            params=params,
                            json=json_body,
                        )
                        response.raise_for_status()
                except TimeoutError as err:
                    msg = f"Timeout error while accessing {method} {url}: {err}"
                    raise PortainerTimeoutError(msg) from err
                except ClientResponseError as err:
                    match err.status:
                        case 401:
                            msg = f"Authentication failed for {method} {url}: Invalid API key"
                            raise PortainerAuthenticationError(msg) from err
                        case 404:
                            msg = f"Resource not found at {method} {url}: {err}"
                            raise PortainerNotFoundError(msg) from err
                        case _:
                            msg = f"Connection error for {method} {url}: {err}"
                            raise PortainerConnectionError(msg) from err
                except (ClientError, socket.gaierror) as err:
                    msg = f"Unexpected error during {method} {url}: {err}"
                    raise PortainerConnectionError(msg) from err

        if response.status in (204, 304):
            return None

        content_type = response.headers.get("Content-Type", "")
        if "application/json" not in content_type:
            text = await response.text()
            msg = "Unexpected content type response from the Portainer API"
            raise PortainerError(
                msg,
                {"Content-Type": content_type, "response": text},
            )

        # Read events instead. Ideal for getting image pull progress
        events: list[Any] = []
        if not parse:
            async for chunk in response.content:
                for line in chunk.splitlines():
                    stripped_line = line.strip()
                    if not stripped_line:
                        continue
                    events.append(json.loads(stripped_line))
            return events

        return await response.json()

    async def _stream_request(
        self,
        uri: str,
        *,
        params: dict[str, Any] | None = None,
    ) -> AsyncGenerator[dict[str, Any], None]:
        """Open a persistent streaming connection and yield JSON events as they arrive.

        Unlike :meth:`_request`, this method does not buffer the full response.
        The connection remains open until cancelled or the server closes it.
        The connection-establishment step is subject to the normal request timeout;
        the ongoing stream is not time-limited.

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

        Yields:
        ------
            Parsed JSON objects, one per newline-delimited event.

        Raises:
        ------
            PortainerTimeoutError: If the connection cannot be established within the timeout.
            PortainerAuthenticationError: If the API key is invalid.
            PortainerConnectionError: On network errors.

        """
        url = URL.build(
            scheme=self._api_scheme,
            host=self._api_host,
            port=self._api_port,
            path=f"{self._api_base_path}/api/",
        ).join(URL(uri))

        headers = {
            "Accept": "application/json, text/plain",
            "User-Agent": f"PythonPortainer/{VERSION}",
            "X-API-Key": self._api_key,
        }

        if self._session is None:
            self._session = ClientSession()
            self._close_session = True

        try:
            async with asyncio.timeout(self._request_timeout):
                response = await self._session.request(
                    METH_GET,
                    url,
                    headers=headers,
                    params=params,
                )
                response.raise_for_status()
        except TimeoutError as err:
            msg = f"Timeout error while connecting to {url}: {err}"
            raise PortainerTimeoutError(msg) from err
        except ClientResponseError as err:
            match err.status:
                case 401:
                    msg = f"Authentication failed for {url}: Invalid API key"
                    raise PortainerAuthenticationError(msg) from err
                case 404:
                    msg = f"Resource not found at {url}: {err}"
                    raise PortainerNotFoundError(msg) from err
                case _:
                    msg = f"Connection error for {url}: {err}"
                    raise PortainerConnectionError(msg) from err
        except (ClientError, socket.gaierror) as err:
            msg = f"Unexpected error connecting to {url}: {err}"
            raise PortainerConnectionError(msg) from err

        try:
            buffer = b""
            async for chunk in response.content:
                buffer += chunk
                while b"\n" in buffer:
                    line, buffer = buffer.split(b"\n", 1)
                    stripped = line.strip()
                    if stripped:
                        yield json.loads(stripped)
        finally:
            response.release()

    async def get_events(
        self,
        endpoint_id: int,
        *,
        since: datetime | None = None,
        until: datetime | None = None,
        filters: dict[str, list[str]] | None = None,
    ) -> AsyncGenerator[DockerEvent, None]:
        """Stream Docker events from an endpoint in real time.

        Opens a persistent connection to the Docker events endpoint and yields
        :class:`~pyportainer.models.docker.DockerEvent` objects as they are emitted.
        When ``until`` is provided the Docker daemon closes the connection once
        all matching events have been sent, making this suitable for bounded
        queries as well as infinite streams.

        Args:
        ----
            endpoint_id: The ID of the Portainer endpoint to stream events from.
            since: Only return events after this timestamp. If timezone-naive,
                UTC is assumed.
            until: Only return events before this timestamp. If timezone-naive,
                UTC is assumed. When supplied, the stream ends automatically.
            filters: Optional Docker event filters, e.g.
                ``{"type": ["container"], "event": ["start", "die"]}``.

        Yields:
        ------
            :class:`~pyportainer.models.docker.DockerEvent` objects.

        """
        params: dict[str, Any] = {}
        if since is not None:
            params["since"] = int(since.timestamp())
        if until is not None:
            params["until"] = int(until.timestamp())
        if filters is not None:
            params["filters"] = json.dumps(filters)

        async for raw in self._stream_request(
            f"endpoints/{endpoint_id}/docker/events",
            params=params or None,
        ):
            yield DockerEvent.from_dict(raw)

    async def get_recent_events(
        self,
        endpoint_id: int,
        *,
        since: datetime,
        until: datetime | None = None,
        filters: dict[str, list[str]] | None = None,
    ) -> list[DockerEvent]:
        """Return Docker events from an endpoint for a bounded time window.

        Unlike :meth:`get_events`, this method collects all matching events into
        a list and returns once the time window is exhausted. It is a thin
        wrapper around :meth:`get_events` that passes ``until`` (defaulting to
        now) so the Docker daemon closes the connection automatically.

        Args:
        ----
            endpoint_id: The ID of the Portainer endpoint to query.
            since: Only return events after this timestamp. If timezone-naive,
                UTC is assumed.
            until: Only return events before this timestamp. Defaults to the
                current UTC time if not provided.
            filters: Optional Docker event filters, e.g.
                ``{"type": ["container"], "event": ["start", "die"]}``.

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

        """
        if until is None:
            until = datetime.now(UTC)

        return [
            event
            async for event in self.get_events(
                endpoint_id,
                since=since,
                until=until,
                filters=filters,
            )
        ]

    async def get_endpoints(self) -> list[Endpoint]:
        """Get the list of endpoints from the Portainer API.

        Returns
        -------
            A list of Endpoint objects.

        """
        endpoints = await self._request("endpoints")

        return [Endpoint.from_dict(endpoint) for endpoint in endpoints]

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

        Args:
        ----
            endpoint_id: The ID of the endpoint to get containers from.
            all: If True, include all containers. If False, only running containers.

        Returns:
        -------
            A list of containers.

        """
        containers = await self._request(f"endpoints/{endpoint_id}/docker/containers/json?all=1")

        return [DockerContainer.from_dict(container) for container in containers]

    async def start_container(self, endpoint_id: int, container_id: str) -> Any:
        """Start a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/start",
            method="POST",
            json_body={},
        )

    async def stop_container(self, endpoint_id: int, container_id: str) -> Any:
        """Stop a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/stop",
            method="POST",
        )

    async def restart_container(self, endpoint_id: int, container_id: str) -> Any:
        """Restart a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/restart",
            method="POST",
        )

    async def pause_container(self, endpoint_id: int, container_id: str) -> Any:
        """Pause a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/pause",
            method="POST",
        )

    async def unpause_container(self, endpoint_id: int, container_id: str) -> Any:
        """Unpause a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/unpause",
            method="POST",
        )

    async def kill_container(self, endpoint_id: int, container_id: str) -> Any:
        """Kill a container on the specified endpoint.

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

        """
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/kill",
            method="POST",
        )

    async def delete_container(self, endpoint_id: int, container_id: str, *, force: bool = False) -> Any:
        """Delete a container on the specified endpoint.

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

        """
        params = {"force": str(force).lower()}
        return await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}",
            method="DELETE",
            params=params,
        )

    async def inspect_container(self, endpoint_id: int, container_id: str, *, raw: bool = False) -> DockerInspect | Any:
        """Inspect a container on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            container_id: The ID of the container to inspect.
            raw: If True, return the raw JSON response. If False, return a DockerInspect object.

        Returns:
        -------
            A DockerContainer object with the inspected data.

        """
        container = await self._request(f"endpoints/{endpoint_id}/docker/containers/{container_id}/json")

        if raw:
            return container
        return DockerInspect.from_dict(container)

    async def docker_version(self, endpoint_id: int) -> DockerVersion:
        """Get the Docker version on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.

        Returns:
        -------
            A DockerVersion object with the Docker version data.

        """
        version = await self._request(f"endpoints/{endpoint_id}/docker/version")

        return DockerVersion.from_dict(version)

    async def docker_info(self, endpoint_id: int) -> DockerInfo:
        """Get the Docker info on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.

        Returns:
        -------
            A DockerInfo object with the Docker info data.

        """
        info = await self._request(f"endpoints/{endpoint_id}/docker/info")

        return DockerInfo.from_dict(info)

    async def container_stats(
        self,
        endpoint_id: int,
        container_id: str,
        *,
        stream: bool = False,
        one_shot: bool = True,
    ) -> Any:
        """Get the stats of a container on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            container_id: The ID of the container to get stats from.
            stream: If True, stream the stats. If False, get a single snapshot.
            one_shot: If True, get a single snapshot. If False, stream the stats.

        Returns:
        -------
            The stats of the container.

        """
        params = {"stream": str(stream).lower(), "one-shot": str(one_shot).lower()}
        stats = await self._request(
            f"endpoints/{endpoint_id}/docker/containers/{container_id}/stats",
            params=params,
        )

        return DockerContainerStats.from_dict(stats)

    async def get_image_information(self, endpoint_id: int, image_id: str) -> ImageInformation:
        """Get information about a Docker image.

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

        Returns:
        -------
            An ImageInformation object with the image data.

        """
        image = await self._request(f"endpoints/{endpoint_id}/docker/distribution/{image_id}/json")

        return ImageInformation.from_dict(image)

    async def get_image(self, endpoint_id: int, image_id: str) -> LocalImageInformation:
        """Get information about a Docker image.

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

        Returns:
        -------
            A LocalImageInformation object with the image data.

        """
        image = await self._request(f"endpoints/{endpoint_id}/docker/images/{image_id}/json")

        return LocalImageInformation.from_dict(image)

    async def container_image_status(self, endpoint_id: int, image: str) -> PortainerImageUpdateStatus:
        """Check whether a newer version of a Docker image is available in the registry.

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

        Returns:
        -------
            A PortainerImageUpdateStatus with the comparison result and digests.

        """
        local, remote = await asyncio.gather(
            self.get_image(endpoint_id, image),
            self.get_image_information(endpoint_id, image),
        )

        registry_digest = remote.descriptor.digest if remote.descriptor else None
        local_digest = next(
            (digest.partition("@")[2] for digest in (local.repo_digests or []) if "@" in digest),
            None,
        )

        return PortainerImageUpdateStatus(
            update_available=bool(registry_digest and registry_digest != local_digest),
            local_digest=local_digest,
            registry_digest=registry_digest,
        )

    async def image_recreate(self, endpoint_id: int, image_id: str, timeout: timedelta = timedelta(minutes=5)) -> Any:
        """Recreate a Docker image.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            image_id: The ID of the image to recreate.
            timeout: Timeout for the image recreation process. Defaults to 5 minutes.

        Returns:
        -------
            An ImageInformation object with the recreated image data.

        """
        params = {"fromImage": image_id}
        return await self._request(
            uri=f"endpoints/{endpoint_id}/docker/images/create?fromImage={image_id}",
            timeout=timeout.total_seconds(),
            method="POST",
            params=params,
            parse=False,
        )

    async def container_recreate_helper(self, endpoint_id: int, container_id: str, image: str, timeout: timedelta = timedelta(minutes=5)) -> Any:
        """Recreate a Docker container service.

        This helper runs through the Portainer API and recreates the specified container.
        It first inspects the container to get its configuration, then creates a new container
        with the same configuration.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            container_id: The ID of the container to recreate.
            image: The tag of the image to use for the new container.

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

        """
        container_inspect = await self.inspect_container(
            endpoint_id=endpoint_id,
            container_id=container_id,
            raw=True,
        )

        if not isinstance(container_inspect, dict):
            msg = "Failed to inspect container for recreation."
            raise PortainerError(msg)

        await self.image_recreate(
            endpoint_id=endpoint_id,
            image_id=image,
            timeout=timeout,
        )

        await self.stop_container(
            endpoint_id=endpoint_id,
            container_id=container_id,
        )

        await self.delete_container(
            endpoint_id=endpoint_id,
            container_id=container_id,
            force=True,
        )

        create_body = {
            **container_inspect["Config"],
            "Image": image,
            "HostConfig": container_inspect["HostConfig"],
            "Config": container_inspect["Config"],
        }

        created = await self.container_create(
            endpoint_id=endpoint_id,
            name=container_inspect.get("Name", "").lstrip("/"),
            image=image,
            config=create_body,
        )

        # This is optional; reattach networks the same way as the original container
        # I have to test this, probablt need some friendly users...
        networks = (container_inspect["NetworkSettings"] or {}).get("Networks") or {}
        for net_name in networks:
            network_name = (container_inspect["HostConfig"] or {}).get("NetworkMode") or ""
            if network_name in {"host", "none"} or network_name.startswith("container:"):
                continue

            await self._request(
                f"endpoints/{endpoint_id}/docker/networks/{net_name}/connect",
                method="POST",
                json_body={"Container": created.id},
            )

        await self.start_container(
            endpoint_id=endpoint_id,
            container_id=created.id,
        )

        return created

    async def container_recreate(
        self, endpoint_id: int, container_id: str, timeout: timedelta = timedelta(minutes=5), *, pull_image: bool = False
    ) -> DockerContainer:
        """Recreate a Docker container.

        Args:
        ----
            endpoint_id: The ID of the endpoint.
            container_id: The ID of the container to recreate.
            timeout: Timeout for the container recreation process. Defaults to 5 minutes.
            pull_image: If True, pull the latest image before recreating the container.

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

        """
        params = {"PullImage": pull_image}
        container = await self._request(
            uri=f"docker/{endpoint_id}/containers/{container_id}/recreate",
            method="POST",
            json_body=params,
            timeout=timeout.total_seconds(),
        )

        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.

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

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

        return DockerImagePruneResponse.from_dict(response)

    async def docker_system_df(self, endpoint_id: int) -> Any:
        """Get Docker system disk usage on the specified endpoint.

        Args:
        ----
            endpoint_id: The ID of the endpoint.

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

        """
        response = await self._request(
            f"endpoints/{endpoint_id}/docker/system/df",
            method="GET",
        )

        return DockerSystemDF.from_dict(response)

    async def portainer_system_status(self) -> PortainerSystemStatus:
        """Get the system status of the Portainer instance.

        Returns
        -------
            A PortainerSystemStatus object with the system status data.

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

        return PortainerSystemStatus.from_dict(status)

    async def get_stacks(
        self,
        endpoint_id: int | None = None,
        swarm_id: str | None = None,
    ) -> list[Stack]:
        """Get the list of stacks from the Portainer API.

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

        Returns:
        -------
            A list of stacks.

        """
        filters: dict[str, Any] = {}
        if endpoint_id is not None:
            filters["EndpointID"] = endpoint_id
        if swarm_id is not None:
            filters["SwarmID"] = swarm_id

        params = filters and {"filters": json.dumps(filters)}
        stacks = await self._request("stacks", params=params)

        if stacks is None:  # 204 response = no stacks
            return []
        return [Stack.from_dict(stack) for stack in stacks]

    async def get_stack(self, stack_id: int) -> Stack:
        """Get details of a specific stack.

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

        Returns:
        -------
            Stack details.

        """
        stack = await self._request(f"stacks/{stack_id}")
        return Stack.from_dict(stack)

    async def get_stack_containers(
        self,
        endpoint_id: int,
        stack_name: str,
    ) -> list[DockerContainer]:
        """Get containers belonging to a stack.

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

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

        Returns:
        -------
            A list of containers in the stack.

        """
        filters = {"label": [f"com.docker.compose.project={stack_name}"]}
        params = {"all": "1", "filters": json.dumps(filters)}
        containers = await self._request(
            f"endpoints/{endpoint_id}/docker/containers/json",
            params=params,
        )
        return [DockerContainer.from_dict(container) for container in containers]

    async def start_stack(self, endpoint_id: int, stack_id: int, timeout: timedelta = timedelta(minutes=5)) -> Stack:
        """Start a stopped stack.

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

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

        """
        stack = await self._request(
            f"stacks/{stack_id}/start",
            method=METH_POST,
            params={"endpointId": endpoint_id},
            timeout=timeout.total_seconds(),
        )
        return Stack.from_dict(stack)

    async def stop_stack(self, endpoint_id: int, stack_id: int, timeout: timedelta = timedelta(minutes=5)) -> Stack:
        """Stop a running stack.

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

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

        """
        stack = await self._request(
            f"stacks/{stack_id}/stop",
            method=METH_POST,
            params={"endpointId": endpoint_id},
            timeout=timeout.total_seconds(),
        )
        return Stack.from_dict(stack)

    async def delete_stack(
        self,
        stack_id: int,
        endpoint_id: int,
        *,
        external: bool = False,
    ) -> None:
        """Delete a stack.

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

        """
        params: dict[str, Any] = {
            "endpointId": endpoint_id,
            "external": str(external).lower(),
        }
        await self._request(
            f"stacks/{stack_id}",
            method=METH_DELETE,
            params=params,
        )

    async def 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, 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 requests (in seconds).
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,
    session: ClientSession | None = None,
    max_retries: int = 3,
) -> None:
    """Initialize the Portainer object.

    Args:
    ----
        api_url: URL of the Portainer API.
        api_key: API key for authentication.
        request_timeout: Timeout for requests (in seconds).
        session: Optional aiohttp session to use.
        max_retries: Maximum number of retry attempts on transient errors.

    """
    self._api_key = api_key
    self._request_timeout = request_timeout
    self._session = session
    self._max_retries = max_retries

    parsed_url = urlparse(api_url)
    self._api_host = parsed_url.hostname or ""
    self._api_scheme = parsed_url.scheme or ""
    self._api_port = parsed_url.port

    self._api_base_path = (parsed_url.path or "").rstrip("/")

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

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

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

Get Docker system disk usage on the specified endpoint.

---

endpoint_id: The ID of the endpoint.

---

The response from the Portainer API.

Source code in src/pyportainer/pyportainer.py

async def docker_system_df(self, endpoint_id: int) -> Any:
    """Get Docker system disk usage on the specified endpoint.

    Args:
    ----
        endpoint_id: The ID of the endpoint.

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

    """
    response = await self._request(
        f"endpoints/{endpoint_id}/docker/system/df",
        method="GET",
    )

    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_containers(endpoint_id) async

Get the list of containers from the Portainer API.

---

endpoint_id: The ID of the endpoint to get containers from.
all: If True, include all containers. If False, only running containers.

---

A list of containers.

Source code in src/pyportainer/pyportainer.py

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

    Args:
    ----
        endpoint_id: The ID of the endpoint to get containers from.
        all: If True, include all containers. If False, only running containers.

    Returns:
    -------
        A list of containers.

    """
    containers = await self._request(f"endpoints/{endpoint_id}/docker/containers/json?all=1")

    return [DockerContainer.from_dict(container) for container in containers]

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.docker.DockerEvent objects as they are emitted. When until is provided the Docker daemon closes the connection once all matching events have been sent, making this suitable for bounded queries as well as infinite streams.

---

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.docker.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.docker.DockerEvent` objects as they are emitted.
    When ``until`` is provided the Docker daemon closes the connection once
    all matching events have been sent, making this suitable for bounded
    queries as well as infinite streams.

    Args:
    ----
        endpoint_id: The ID of the Portainer endpoint to stream events from.
        since: Only return events after this timestamp. If timezone-naive,
            UTC is assumed.
        until: Only return events before this timestamp. If timezone-naive,
            UTC is assumed. When supplied, the stream ends automatically.
        filters: Optional Docker event filters, e.g.
            ``{"type": ["container"], "event": ["start", "die"]}``.

    Yields:
    ------
        :class:`~pyportainer.models.docker.DockerEvent` objects.

    """
    params: dict[str, Any] = {}
    if since is not None:
        params["since"] = int(since.timestamp())
    if until is not None:
        params["until"] = int(until.timestamp())
    if filters is not None:
        params["filters"] = json.dumps(filters)

    async for raw in self._stream_request(
        f"endpoints/{endpoint_id}/docker/events",
        params=params or None,
    ):
        yield DockerEvent.from_dict(raw)

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.docker.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.docker.DockerEvent` objects,
        ordered as received from the Docker daemon.

    """
    if until is None:
        until = datetime.now(UTC)

    return [
        event
        async for event in self.get_events(
            endpoint_id,
            since=since,
            until=until,
            filters=filters,
        )
    ]

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_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]

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?fromImage={image_id}",
        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.

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

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

    return DockerImagePruneResponse.from_dict(response)

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)

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)

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

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.

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

    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),
        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 reconnecting after a
                dropped connection. Defaults to 5 seconds.
            debug: Enable debug logging.

        """
        self._portainer = portainer
        self._endpoint_id = endpoint_id
        self._event_types = event_types
        self._reconnect_interval = reconnect_interval
        self._task: asyncio.Task[None] | None = None
        self._callbacks: list[EventListenerCallback] = []

        _LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)

    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 _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):
            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 ``reconnect_interval``.

        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:
                _LOGGER.warning(
                    "Timeout on endpoint %s, reconnecting in %ss",
                    endpoint_id,
                    self._reconnect_interval.total_seconds(),
                )
            except PortainerConnectionError:
                _LOGGER.warning(
                    "Connection lost on endpoint %s, reconnecting in %ss",
                    endpoint_id,
                    self._reconnect_interval.total_seconds(),
                )
            except PortainerError:
                _LOGGER.exception(
                    "Error on endpoint %s, reconnecting in %ss",
                    endpoint_id,
                    self._reconnect_interval.total_seconds(),
                )

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

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

        Runs all per-endpoint listeners concurrently via :func:`asyncio.gather`.
        """
        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 listen to.")
            endpoints = await self._portainer.get_endpoints()
            endpoint_ids = [endpoint.id for endpoint in endpoints]

        await asyncio.gather(*(self._listen_with_reconnect(ep_id) for ep_id in endpoint_ids))

__init__(portainer, endpoint_id=None, *, event_types=None, reconnect_interval=timedelta(seconds=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 reconnecting after a
    dropped connection. Defaults to 5 seconds.
debug: Enable debug logging.

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),
    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 reconnecting after a
            dropped connection. Defaults to 5 seconds.
        debug: Enable debug logging.

    """
    self._portainer = portainer
    self._endpoint_id = endpoint_id
    self._event_types = event_types
    self._reconnect_interval = reconnect_interval
    self._task: asyncio.Task[None] | None = None
    self._callbacks: list[EventListenerCallback] = []

    _LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)

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

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.

        """
        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] = []

        _LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)

    @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.

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.

    """
    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] = []

    _LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)

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."""