Source code for podman.domain.containers_manager

"""PodmanResource manager subclassed for Containers."""

import logging
import urllib
from collections.abc import Mapping
from typing import Any, Union

from podman import api
from podman.domain.containers import Container
from podman.domain.containers_create import CreateMixin
from podman.domain.containers_run import RunMixin
from podman.domain.manager import Manager
from podman.errors import APIError

logger = logging.getLogger("podman.containers")




class ContainersManager(RunMixin, CreateMixin, Manager):
    """Specialized Manager for Container resources."""

    @property
    def resource(self):
        """Type[Container]: prepare_model() will create Container classes."""
        return Container



    def exists(self, key: str) -> bool:
        response = self.client.get(f"/containers/{key}/exists")
        return response.ok




    def get(self, key: str, **kwargs) -> Container:
        """Get container by name or id.

        Args:
            key: Container name or id.

        Keyword Args:
            compatible (bool): Use Docker compatibility endpoint

        Returns:
            A `Container` object corresponding to `key`.

        Raises:
            NotFound: when Container does not exist
            APIError: when an error return by service
        """
        compatible = kwargs.get("compatible", False)

        container_id = urllib.parse.quote_plus(key)
        response = self.client.get(f"/containers/{container_id}/json", compatible=compatible)
        response.raise_for_status()
        return self.prepare_model(attrs=response.json())




    def list(self, **kwargs) -> list[Container]:
        """Report on containers.

        Keyword Args:
            all: If False, only show running containers. Default: False.
            since: Show containers created after container name or id given.
            before: Show containers created before container name or id given.
            limit: Show last N created containers.
            filters: Filter container reported.
                Available filters:

                - exited (int): Only containers with specified exit code
                - status (str): One of restarting, running, paused, exited
                - label (Union[str, list[str]]): Format either "key", "key=value" or a list of such.
                - id (str): The id of the container.
                - name (str): The name of the container.
                - ancestor (str): Filter by container ancestor. Format of
                    <image-name>[:tag], <image-id>, or <image@digest>.
                - before (str): Only containers created before a particular container.
                    Give the container name or id.
                - since (str): Only containers created after a particular container.
                    Give container name or id.
            sparse: If False, return basic container information without additional
                inspection requests. This improves performance when listing many containers
                but might provide less detail. You can call Container.reload() on individual
                containers later to retrieve complete attributes. Default: True.
                When Docker compatibility is enabled with `compatible=True`: Default: False.
            ignore_removed: If True, ignore failures due to missing containers.

        Raises:
            APIError: when service returns an error
        """
        compatible = kwargs.get("compatible", False)

        # Set sparse default based on mode:
        # Libpod behavior: default is sparse=True (faster, requires reload for full details)
        # Docker behavior: default is sparse=False (full details immediately, compatible)
        if "sparse" in kwargs:
            sparse = kwargs["sparse"]
        else:
            sparse = not compatible  # True for libpod, False for compat

        params = {
            "all": kwargs.get("all"),
            "filters": kwargs.get("filters", {}),
            "limit": kwargs.get("limit"),
        }
        if "before" in kwargs:
            params["filters"]["before"] = kwargs.get("before")
        if "since" in kwargs:
            params["filters"]["since"] = kwargs.get("since")

        # filters formatted last because some kwargs may need to be mapped into filters
        params["filters"] = api.prepare_filters(params["filters"])

        response = self.client.get("/containers/json", params=params, compatible=compatible)
        response.raise_for_status()

        containers: list[Container] = [self.prepare_model(attrs=i) for i in response.json()]

        # If sparse is False, reload each container to get full details
        if not sparse:
            for container in containers:
                try:
                    container.reload(compatible=compatible)
                except APIError:
                    # Skip containers that might have been removed
                    pass

        return containers




    def prune(self, filters: Mapping[str, str] = None) -> dict[str, Any]:
        """Delete stopped containers.

        Args:
            filters: Criteria for determining containers to remove. Available keys are:
                - until (str): Delete containers before this time
                - label (list[str]): Labels associated with containers

        Returns:
            Keys:
                - ContainersDeleted (list[str]): Identifiers of deleted containers.
                - SpaceReclaimed (int): Amount of disk space reclaimed in bytes.

        Raises:
            APIError: when service reports an error
        """
        params = {"filters": api.prepare_filters(filters)}
        response = self.client.post("/containers/prune", params=params)
        response.raise_for_status()

        results = {"ContainersDeleted": [], "SpaceReclaimed": 0}
        for entry in response.json():
            if entry.get("Err") is not None:
                raise APIError(
                    entry["Err"],
                    response=response,
                    explanation=f"""Failed to prune container '{entry["Id"]}'""",
                )

            results["ContainersDeleted"].append(entry["Id"])  # type: ignore[attr-defined]
            results["SpaceReclaimed"] += entry["Size"]
        return results




    def remove(self, container_id: Union[Container, str], **kwargs):
        """Delete container.

        Podman only

        Args:
            container_id: identifier of Container to delete.

        Keyword Args:
            v (bool): Delete associated volumes as well.
            link (bool): Ignored.
            force (bool): Kill a running container before deleting.
        """
        if isinstance(container_id, Container):
            container_id = container_id.id

        # v is used for the compat endpoint while volumes is used for the libpod endpoint
        params = {"v": kwargs.get("v"), "force": kwargs.get("force"), "volumes": kwargs.get("v")}

        response = self.client.delete(f"/containers/{container_id}", params=params)
        response.raise_for_status()