Source code for podman.domain.manager

"""Base classes for PodmanResources and Manager's."""

from abc import ABC, abstractmethod
from collections import abc
from typing import Any, List, Mapping, Optional, TypeVar, Union

from podman.api.client import APIClient

# Methods use this Type when a subclass of PodmanResource is expected.
PodmanResourceType: TypeVar = TypeVar("PodmanResourceType", bound="PodmanResource")

[docs] class PodmanResource(ABC): """Base class for representing resource of a Podman service. Attributes: attrs: Mapping of attributes for resource from Podman service """ def __init__( self, attrs: Optional[Mapping[str, Any]] = None, client: Optional[APIClient] = None, collection: Optional["Manager"] = None, ): """Initialize base class for PodmanResource's. Args: attrs: Mapping of attributes for resource from Podman service. client: Configured connection to a Podman service. collection: Manager of this category of resource, named `collection` for compatibility """ super().__init__() self.client = client self.manager = collection self.attrs = {} if attrs is not None: self.attrs.update(attrs) def __repr__(self): return f"<{self.__class__.__name__}: {self.short_id}>" def __eq__(self, other): return isinstance(other, self.__class__) and == def __hash__(self): return hash(f"{self.__class__.__name__}:{}") @property def id(self): # pylint: disable=invalid-name """str: Returns the identifier for the object.""" return self.attrs.get("Id") @property def short_id(self): """str: Returns truncated identifier. 'sha256' preserved when included in the id. No attempt is made to ensure the returned value is semantically meaningful for all resources. """ if"sha256:"): return[:17] return[:10]
[docs] def reload(self) -> None: """Refresh this object's data from the service.""" latest = self.manager.get( self.attrs = latest.attrs
[docs] class Manager(ABC): """Base class for representing a Manager of resources for a Podman service.""" @property @abstractmethod def resource(self): """Type[PodmanResource]: Class which the factory method prepare_model() will use.""" def __init__(self, client: APIClient = None) -> None: """Initialize Manager() object. Args: client: APIClient() configured to connect to Podman service. """ super().__init__() self.client = client
[docs] @abstractmethod def exists(self, key: str) -> bool: """Returns True if resource exists. Podman only. Notes: This method does _not_ provide any mutex mechanism. """
[docs] @abstractmethod def get(self, key: str) -> PodmanResourceType: """Returns representation of resource."""
[docs] @abstractmethod def list(self, **kwargs) -> List[PodmanResourceType]: """Returns list of resources."""
[docs] def prepare_model(self, attrs: Union[PodmanResource, Mapping[str, Any]]) -> PodmanResourceType: """Create a model from a set of attributes.""" # Refresh existing PodmanResource. if isinstance(attrs, PodmanResource): attrs.client = self.client attrs.collection = self return attrs # Instantiate new PodmanResource from Mapping[str, Any] if isinstance(attrs, abc.Mapping): # TODO Determine why pylint is reporting typing.Type not callable # pylint: disable=not-callable return self.resource(attrs=attrs, client=self.client, collection=self) # pylint: disable=broad-exception-raised raise Exception(f"Can't create {self.resource.__name__} from {attrs}")