1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
"""Base classes for PodmanResources and Manager's."""
from abc import ABC, abstractmethod
from collections import abc
from typing import Any, Optional, TypeVar, Union
from collections.abc import Mapping
from podman.api.client import APIClient
# Methods use this Type when a subclass of PodmanResource is expected.
PodmanResourceType: TypeVar = TypeVar("PodmanResourceType", bound="PodmanResource")
class PodmanResource(ABC): # noqa: B024
"""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,
podman_client: Optional["PodmanClient"] = 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
podman_client: PodmanClient() configured to connect to Podman object.
"""
super().__init__()
self.client = client
self.manager = collection
self.podman_client = podman_client
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 self.id == other.id
def __hash__(self):
return hash(f"{self.__class__.__name__}:{self.id}")
@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 self.id.startswith("sha256:"):
return self.id[:17]
return self.id[:10]
def reload(self) -> None:
"""Refresh this object's data from the service."""
latest = self.manager.get(self.id)
self.attrs = latest.attrs
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: Optional[APIClient] = None, podman_client: Optional["PodmanClient"] = None
) -> None:
"""Initialize Manager() object.
Args:
client: APIClient() configured to connect to Podman service.
podman_client: PodmanClient() configured to connect to Podman object.
"""
super().__init__()
self.client = client
self.podman_client = podman_client
@abstractmethod
def exists(self, key: str) -> bool:
"""Returns True if resource exists.
Podman only.
Notes:
This method does _not_ provide any mutex mechanism.
"""
@abstractmethod
def get(self, key: str) -> PodmanResourceType:
"""Returns representation of resource."""
@abstractmethod
def list(self, **kwargs) -> list[PodmanResourceType]:
"""Returns list of resources."""
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.podman_client = self.podman_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, podman_client=self.podman_client, collection=self
)
# pylint: disable=broad-exception-raised
raise Exception(f"Can't create {self.resource.__name__} from {attrs}")
|
