Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def query_model(model, identifiers=None):
"""Retrieve all (default) or given model(s)."""
objs = []
if identifiers is None:
objs.extend(model.list())
else:
try:
for ident in identifiers:
objs.append(model.get(ident))
except (
podman.PodNotFound,
podman.ImageNotFound,
podman.ContainerNotFound,
) as ex:
print(
'"{}" not found'.format(ex.name), file=sys.stderr, flush=True)
return objs
def pull(self):
"""Store image elsewhere."""
try:
try:
img = self.client.images.get(self._args.image[0])
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
else:
img.push(self._args.tag[0], tlsverify=self._args.tlsverify)
print(self._args.image[0])
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
def create(self):
"""Create container."""
try:
for ident in self._args.image:
try:
img = self.client.images.get(ident)
img.container(**self.opts)
print(ident)
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
def attach(self):
"""Attach to instantiated image."""
args = {
'detach': True,
'tty': True,
}
if self._args.command:
args['command'] = self._args.command
try:
try:
ident = self.client.images.pull(self._args.image)
img = self.client.images.get(ident)
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
return 1
ctnr = img.create(**args)
ctnr.attach(eot=4)
try:
ctnr.start()
print()
except (BrokenPipeError, KeyboardInterrupt):
print('\nContainer disconnected.')
except podman.ErrorOccurred as e:
def pull(self):
"""Retrieve image."""
for ident in self._args.targets:
try:
self.client.images.pull(ident)
print(ident)
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)
def run(self):
"""Run container."""
for ident in self._args.image:
try:
try:
img = self.client.images.get(ident)
ctnr = img.container(**self.opts)
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
continue
else:
logging.debug('New container created "{}"'.format(ctnr.id))
if self._args.detach:
ctnr.start()
print(ctnr.id)
else:
ctnr.attach(eot=4)
ctnr.start()
print(ctnr.id)
def _get_image(self, ident):
try:
logging.debug("Getting image %s", ident)
img = self.client.images.get(ident)
except podman.ImageNotFound:
pass
else:
return img.inspect()
def remove(self):
"""Remove image(s)."""
for ident in self._args.targets:
try:
img = self.client.images.get(ident)
img.remove(self._args.force)
print(ident)
except podman.ImageNotFound as e:
sys.stdout.flush()
print(
'Image {} not found.'.format(e.name),
file=sys.stderr,
flush=True)
except podman.ErrorOccurred as e:
sys.stdout.flush()
print(
'{}'.format(e.reason).capitalize(),
file=sys.stderr,
flush=True)