import logging
import uuid
from collections import OrderedDict
from time import sleep
from docker import errors
from docker.helpers import execute
logger = logging.getLogger(__name__)
[docs]class Docker(object):
"""
Docker manager which can start and stop containers in addition to run commands with docker exec.
The manager also have a few helper functions for things like listing files and directories.
"""
def __init__(self, image='ubuntu', name_prefix='dyn', timeout=3600, privilege=False,
combine_outputs=False, env_variables=None, ports_mapping=None):
"""
Creates a docker manager. Each manager has a reference to a unique container name.
:param image: The image which the manager should use to start the container. It could be
a local image or an image from the registry.
:type image: str
:param name_prefix: All image names are prefixed with this string.
:type name_prefix: str
:param timeout: The time the docker container will live after running ``docker.start()`` in
seconds.
:type timeout: int
:param privilege: If set to True the docker container will be run with the privilege
parameter.
:type privilege: bool
:param combine_outputs: Setting this to True will put stderr output in stdout.
:type combine_outputs: bool
:param ports_mapping: Map ports from docker container to host machine,
format ['4080:40480', '5000:5000']
:type ports_mapping: list
:return: A docker manager object.
:rtype: Docker
"""
self.container_name = '{0}-{1}'.format(name_prefix, uuid.uuid4())
self.timeout = timeout
self.image = image
self.privilege = privilege
self.combine_outputs = combine_outputs
self.env_variables = OrderedDict()
if env_variables:
self.env_variables.update(sorted(env_variables.items(), key=lambda t: t[0]))
self.ports = ''
if ports_mapping:
self.ports = ' '.join(['-p {0}'.format(port_mapping) for port_mapping in ports_mapping])
def __enter__(self):
return self.start()
def __exit__(self, exc_type, exc_value, exc_traceback):
self.stop()
if exc_value:
raise exc_value
[docs] def run(self, command, working_directory='', stdin='', login=False, tty=False):
"""
Runs the command with docker exec in the given working directory.
:param command: The command that should be run with docker exec. The command will be wrapped
in `bash -c \'command\'".
:type command: str
:param working_directory: The path to the directory where the command should be run. This
will be evaluated with ``_get_working_directory``, thus relative
paths will become absolute paths.
:type working_directory: str
:type stdin: str
:param login: Will add --login on the bash call.
:type login: boolean
:param tty: Will add -t on the bash call.
:type tty: boolean
:return: A ProcessResult object containing information on the result of the command.
:rtype: ProcessResult
"""
working_directory = self._get_working_directory(working_directory)
command = command.replace('\'', '"')
command_string = 'cd {working_directory} && {envs} {command}'
if self.combine_outputs:
command_string += ' 2>&1'
env_string = ' '.join([
'{0}={1}'.format(key, self.env_variables[key]) for key in self.env_variables
])
result = execute(
'docker exec -i{tty} {container} bash{login} -c \'{command}\''.format(
envs=env_string,
container=self.container_name,
login=' --login' if login else '',
tty=' -t' if tty else '',
command=command_string.format(
working_directory=working_directory,
command=command,
envs=env_string
)
),
stdin
)
return result
[docs] def read_file(self, path):
"""
Reads the content of the file on the given path. Returns None if the file does not exist.
:param path: The path to the file.
:type path: str
:return: The content of the file
:rtype: str
:raises DockerFileNotFoundError: If given an invalid path
:raises DockerWrapperBaseError: For other errors
"""
path = self._get_working_directory(path)
result = self.run('cat {0}'.format(path))
if not result.succeeded:
if errors.FILE_NOT_FOUND_PREDICATE in result.err:
raise errors.DockerFileNotFoundError(path)
raise errors.DockerWrapperBaseError(result.err)
return result.out
[docs] def write_file(self, path, content, append=False):
"""
Write the given content to path.
Overwrites the file if append is set to False.
:param path: The path to the file.
:type path: str
:param content: The content of the file.
:type content: str
:param append: Set to False to overwrite file, defaults to False.
:type append: bool
:return: A object with the result of the create command.
:rtype: ProcessResult
"""
path = self._get_working_directory(path)
modifier = '>>' if append else '>'
return self.run('cat {0} {1}'.format(modifier, path), stdin=content)
[docs] def file_exist(self, path):
"""
Checks whether a file exists or not.
:param path: The path to the file.
:type path: str
:rtype: bool
"""
path = self._get_working_directory(path)
return self.run('test -f {0}'.format(path)).return_code == 0
[docs] def directory_exist(self, path):
"""
Checks whether a directory exists or not.
:param path: The path to the directory.
:type path: str
:rtype: bool
"""
path = self._get_working_directory(path)
return self.run('test -d {0}'.format(path)).return_code == 0
[docs] def list_files(self, path, include_hidden=False):
"""
List files on a given path.
:param path: The path to the directory.
:type path: str
:param include_hidden: Include hidden files in output
:type include_hidden: bool
:return: An list of file names
:rtype: list
:raises DockerFileNotFoundError: If given an invalid path
:raises DockerWrapperBaseError: For other errors
"""
path = self._get_working_directory(path)
# Ignore dot files (hidden files) if include_hidden is enabled:
predicate = '' if include_hidden else '-not -path "*/\.*"'
# The printf part turns './file' into 'file':
result = self.run('find . {0} -maxdepth 1 -type f -printf "%P\n"'.format(predicate), path)
if not result.succeeded:
if errors.FILE_NOT_FOUND_PREDICATE in result.err:
raise errors.DockerFileNotFoundError(path)
raise errors.DockerWrapperBaseError(result.err)
out = result.out.strip()
return sorted(out.split('\n')) if out else []
[docs] def list_directories(self, path, include_trailing_slash=True):
"""
List directories on a given path.
:param path: The path to the directory.
:type path: str
:return: An list of directory names
:rtype: list
:raises DockerFileNotFoundError: If given an invalid path
:raises DockerWrapperBaseError: For other errors
"""
files = []
path = self._get_working_directory(path)
result = self.run('ls -dm */', path)
if not result.succeeded:
if errors.FILE_NOT_FOUND_PREDICATE in result.err:
raise errors.DockerFileNotFoundError(path)
raise errors.DockerWrapperBaseError(result.err)
for file_path in result.out.strip().split(', '):
if include_trailing_slash:
files.append(file_path)
else:
files.append(file_path[:-1])
return files
[docs] def start(self):
"""
Starts a container based on the parameters passed to __init__.
:return: The docker object
"""
if self.privilege:
command_string = 'docker run -d --privileged {0} --name {1} {2} /bin/sleep {3}'
else:
command_string = 'docker run -d {0} --name {1} {2} /bin/sleep {3}'
result = execute(command_string.format(
self.ports,
self.container_name,
self.image,
self.timeout
))
if not result.succeeded:
raise errors.DockerUnavailableError(
'Starting the docker container failed.\n{0}'.format(result.err)
)
return self
[docs] def stop(self):
"""
Stops the container started by this class instance.
:return: The docker object
"""
sleep(2)
execute('docker kill {0}'.format(self.container_name))
execute('docker rm {0}'.format(self.container_name))
return self
@staticmethod
[docs] def wrap(*wrap_args, **wrap_kwargs):
"""
Decorator that wraps the function call in a Docker with statement. It
accepts the same arguments. This decorator adds a docker manager instance
to the kwargs passed into the decorated function.
:return: The decorated function.
"""
def activate(func):
def wrapper(*args, **kwargs):
with Docker(*wrap_args, **wrap_kwargs) as docker:
kwargs['docker'] = docker
return func(*args, **kwargs)
return wrapper
return activate
@staticmethod
def _get_working_directory(working_directory):
"""
Gets the path of the working working directory. It takes a path and converts it to an
appropriate absolute path.
:param working_directory:
:type working_directory: str
:return: An absolute path to the given working directory.
:rtype str:
"""
if working_directory.startswith('/') or working_directory.startswith('~/'):
return working_directory
return '~/{}'.format(working_directory)