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
|
"""Model and Manager for Pod resources."""
import logging
from typing import Any, Optional, Union
from podman.domain.manager import PodmanResource
_Timeout = Union[None, float, tuple[float, float], tuple[float, None]]
logger = logging.getLogger("podman.pods")
class Pod(PodmanResource):
"""Details and configuration for a pod managed by the Podman service."""
@property
def id(self): # pylint: disable=invalid-name
return self.attrs.get("ID", self.attrs.get("Id"))
@property
def name(self):
"""str: Returns name of pod."""
return self.attrs.get("Name")
def kill(self, signal: Union[str, int, None] = None) -> None:
"""Send signal to pod.
Args:
signal: To be sent to pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
response = self.client.post(f"/pods/{self.id}/kill", params={"signal": signal})
response.raise_for_status()
def pause(self) -> None:
"""Pause pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
response = self.client.post(f"/pods/{self.id}/pause")
response.raise_for_status()
def remove(self, force: Optional[bool] = None) -> None:
"""Delete pod.
Args:
force: When True, stop and delete all containers in pod before deleting pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
self.manager.remove(self.id, force=force)
def restart(self) -> None:
"""Restart pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
response = self.client.post(f"/pods/{self.id}/restart")
response.raise_for_status()
def start(self) -> None:
"""Start pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
response = self.client.post(f"/pods/{self.id}/start")
response.raise_for_status()
def stop(self, timeout: _Timeout = None) -> None:
"""Stop pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
params = {"t": timeout}
response = self.client.post(f"/pods/{self.id}/stop", params=params)
response.raise_for_status()
def top(self, **kwargs) -> dict[str, Any]:
"""Report on running processes in pod.
Keyword Args:
ps_args (str): Optional arguments passed to ps.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
params = {
"ps_args": kwargs.get("ps_args"),
"stream": False,
}
response = self.client.get(f"/pods/{self.id}/top", params=params)
response.raise_for_status()
if len(response.text) == 0:
return {"Processes": [], "Titles": []}
return response.json()
def unpause(self) -> None:
"""Unpause pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
response = self.client.post(f"/pods/{self.id}/unpause")
response.raise_for_status()
|
