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
|
"""Model and Manager for Event resources."""
import json
import logging
from datetime import datetime
from typing import Any, Optional, Union
from collections.abc import Iterator
from podman import api
from podman.api.client import APIClient
logger = logging.getLogger("podman.events")
class EventsManager: # pylint: disable=too-few-public-methods
"""Specialized Manager for Event resources."""
def __init__(self, client: APIClient) -> None:
"""Initialize EventManager object.
Args:
client: Connection to Podman service.
"""
self.client = client
def list(
self,
since: Union[datetime, int, None] = None,
until: Union[datetime, int, None] = None,
filters: Optional[dict[str, Any]] = None,
decode: bool = False,
) -> Iterator[Union[str, dict[str, Any]]]:
"""Report on networks.
Args:
decode: When True, decode stream into dict's. Default: False
filters: Criteria for including events.
since: Get events newer than this time.
until: Get events older than this time.
Yields:
When decode is True, Iterator[dict[str, Any]]
When decode is False, Iterator[str]
"""
params = {
"filters": api.prepare_filters(filters),
"since": api.prepare_timestamp(since),
"stream": True,
"until": api.prepare_timestamp(until),
}
response = self.client.get("/events", params=params, stream=True)
response.raise_for_status()
for item in response.iter_lines():
if decode:
yield json.loads(item)
else:
yield item
|
