Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
try:
yield kernel_info
finally:
if kernel_info['id'] in agent.container_registry:
# Container id may be changed (e.g. restarting kernel), so we
# should not rely on the initial value of the container_id.
container_info = agent.container_registry[kernel_info['id']]
container_id = container_info['container_id']
else:
# If fallback to initial container_id if kernel is deleted.
container_id = kernel_info['container_id']
try:
container = docker.containers.container(container_id)
cinfo = await container.show() if container else None
except aiodocker.exceptions.DockerError:
cinfo = None
if cinfo and cinfo['State']['Status'] != 'removing':
await container.delete(force=True)
kernel_ids.append(kid)
info = await agent.create_kernel(kid, config)
container_ids.append(info['container_id'])
# 2 containers are created
assert docker.containers.container(container_ids[0])
assert docker.containers.container(container_ids[1])
await agent.reset()
# Containers are destroyed
with pytest.raises(aiodocker.exceptions.DockerError):
c1 = docker.containers.container(container_ids[0])
c1info = await c1.show()
if c1info['State']['Status'] == 'removing':
raise aiodocker.exceptions.DockerError(
404, {'message': 'success'})
with pytest.raises(aiodocker.exceptions.DockerError):
c2 = docker.containers.container(container_ids[1])
c2info = await c2.show()
if c2info['State']['Status'] == 'removing':
raise aiodocker.exceptions.DockerError(
404, {'message': 'success'})
finally:
for cid in container_ids:
try:
container = docker.containers.container(cid)
cinfo = await container.show() if container else None
except aiodocker.exceptions.DockerError:
cinfo = None
if cinfo and cinfo['State']['Status'] != 'removing':
await container.delete(force=True)
c = await self.client.containers.create(config=config)
await c.put_archive(str(self.source_file_path.parent),
self.create_archive(source))
await c.start()
exit_code = (await c.wait())["StatusCode"]
stdout = "".join(await c.log(stdout=True, stderr=False)).strip()
stderr = "".join(await c.log(stdout=False, stderr=True)).strip()
await c.delete()
duration = int(1000 * (time.time() - start_time))
logger.info("finished in %d ms", duration)
return Result(
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
duration=duration)
except aiodocker.exceptions.DockerError as e:
logger.error(f"docker api error: {e}")
# TODO: better error handling
return None
async def get_container(self, container_name):
"""
Gets the container object having specified name
:param container_name (str): name of the target container
:return: container (object)
"""
container = None
try:
container = await self.docker_client.containers.get(container=container_name)
except aiodocker.exceptions.DockerError as server_error:
self.logger.exception('Error while fetching %s container %s', container_name, server_error)
return container