blob: 89fcf5c153159ba8480978934f21503243203b60 (
plain)
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
|
"""A client for communicating with a Podman varlink service."""
import contextlib
import functools
from varlink import Client as VarlinkClient
from varlink import VarlinkError
from .libs import cached_property
from .libs.containers import Containers
from .libs.errors import error_factory
from .libs.images import Images
from .libs.system import System
class Client(object):
"""A client for communicating with a Podman varlink service.
Example:
>>> import podman
>>> c = podman.Client()
>>> c.system.versions
"""
# TODO: Port to contextlib.AbstractContextManager once
# Python >=3.6 required
def __init__(self,
uri='unix:/run/podman/io.projectatomic.podman',
interface='io.projectatomic.podman'):
"""Construct a podman varlink Client.
uri from default systemd unit file.
interface from io.projectatomic.podman.varlink, do not change unless
you are a varlink guru.
"""
self._podman = None
@contextlib.contextmanager
def _podman(uri, interface):
"""Context manager for API workers to access varlink."""
client = VarlinkClient(address=uri)
try:
iface = client.open(interface)
yield iface
except VarlinkError as e:
raise error_factory(e) from e
finally:
if hasattr(client, 'close'):
client.close()
iface.close()
self._client = functools.partial(_podman, uri, interface)
# Quick validation of connection data provided
try:
if not System(self._client).ping():
raise ValueError('Failed varlink connection "{}/{}"'.format(
uri, interface))
except FileNotFoundError:
raise ValueError('Failed varlink connection "{}/{}".'
' Is podman service running?'.format(
uri, interface))
def __enter__(self):
"""Return `self` upon entering the runtime context."""
return self
def __exit__(self, exc_type, exc_value, traceback):
"""Raise any exception triggered within the runtime context."""
return None
@cached_property
def system(self):
"""Manage system model for podman."""
return System(self._client)
@cached_property
def images(self):
"""Manage images model for libpod."""
return Images(self._client)
@cached_property
def containers(self):
"""Manage containers model for libpod."""
return Containers(self._client)
|
