import docker
from docker.types import Mount
[docs]class Docker:
"""Various utilities to interface with Docker"""
def __init__(self):
self.client = docker.from_env()
self.api = docker.APIClient(base_url="unix://var/run/docker.sock")
[docs] def network_exists(self, name):
"""Return whether the internal network called ``name`` exists
:param name: the network name to search for
:type name: str
:return: a boolean representing whether the network exists
"""
return name in [net.name for net in self.client.networks.list()]
[docs] def create_network(self, name):
"""Return an internal network called ``name``, creating it if
it doesn't already exist
:param name: the name of the network to create
:type name: str
:return: the created :class:`~docker.models.networks.Network`
"""
if self.network_exists(name):
return self.get_network(name)
return self.client.networks.create(name)
[docs] def get_network(self, name, low_level=False):
"""Return the requested network, or its details
:param name: the name of the network to return
:type name: str
:param low_level: flag to return a details dictionary\
instead of a :class:`~docker.models.networks.Network`\
object
:return: the requested :class:`~docker.models.networks.Network`,\
or a dictionary representing it
"""
for net in self.client.networks.list():
if net.name == name:
if low_level:
return self.api.inspect_network(net.id)
return net
[docs] def make_mount(self, host, internal):
"""Create a mount object representing a binding between ``host``
and ``internal``
:param host: the path on the host machine to bind
:type host: str
:param internal: the path inside the container to bind to
:type internal: str
:return: a relevant :class:`~docker.types.Mount` object
"""
return Mount(internal, host, type="bind")
[docs] def image_exists(self, name):
"""Return whether the image tagged ``name`` exists
:param name: the image tag to search for
:type name: str
.. note::
If ``name`` does not contain a tag, the ``:latest``
tag is automatically used.
:return: a boolean representing whether the image exists
"""
if not ":" in name:
name = name + ":latest"
for img in self.client.images.list():
if name in img.tags:
return True
return False
[docs] def pull_image(self, image, tag=None):
"""Pull and save a Docker image by name or URL
:param image: the name or url of the image to pull
:type image: str
:param tag: the initial tag to assign to the image (defaults to ``None``,\
which assigns that tag ``latest``)
:type tag: str
"""
self.client.images.pull(image, tag=tag)
[docs] def pull_image_stream(self, image, tag=None):
"""Pull and save a Docker image by name or URL and stream the output
:param image: the name or url of the image to pull
:type image: str
:param tag: the initial tag to assign to the image (defaults to ``None``,\
which assigns that tag ``latest``)
:type tag: str
:yields: dictionaries representing each streamed output line
"""
yield from self.api.pull(image, tag, stream=True, decode=True)
[docs] def build_image(self, **options):
"""Build a Docker image using the given options
:param options: options to use to the build the image (ideally contains\
at least a path to a build context, as well as a Dockerfile)
:type options: kwargs
"""
self.client.images.build(**options)
[docs] def build_image_stream(self, **options):
"""Build a Docker image using the given options and stream the output
:param options: options to use to the build the image (ideally contains\
at least a path to a build context, as well as a Dockerfile)
:type options: kwargs
:yields: dictionaries representing each streamed output line
"""
yield from self.api.build(decode=True, **options)
[docs] def prune_images(self):
"""Remove all dangling images"""
self.client.images.prune()
[docs] def container_exists(self, name):
"""Return whether the container called ``name`` exists
:param name: the container name to search for
:type name: str
:return: a boolean representing whether the container exists
"""
return name in [con.name for con in self.client.containers.list(all=True)]
[docs] def run_image(self, img, name, **options):
"""Run the requested image as a container
:param img: the name of the image to run
:type img: str
:param name: the name to assign to the target container
:type name: str
:param options: additional options to pass into Docker
:type options: kwargs
:return: the created :class:`~docker.models.containers.Container`
"""
return self.client.containers.run(img, name=name, **options)
[docs] def start_container(self, name):
"""Start the requested container, if it is not running
:param name: the name of the container to start
:type name: str
:return: whether the container was started successfully
"""
if self.container_exists(name):
con = self.client.containers.get(name)
if con.status == "exited":
con.start()
return True
return False
[docs] def stop_container(self, name):
"""Stop the requested container, if it is not stopped
:param name: the name of the container to stop
:type name: str
:return: whether the container was stopped successfully
"""
if self.container_exists(name):
con = self.client.containers.get(name)
if con.status != "exited":
con.kill()
return True
return False
[docs] def wipe_container(self, name):
"""Kill and remove the container named ``name``, if needed,
and return it if it existed
:param name: the name of the container to act on
:type name: str
:return: the removed :class:`~docker.models.containers.Container`\
if it existed, else the string "not found"
"""
for con in self.client.containers.list(all=True):
if name == con.name:
if con.status == "running":
con.kill()
con.remove()
return name
return "not found"
[docs] def exec_command(self, con, command, **options):
"""Run a command on the given container
:param con: the (name of the) container to run the command on
:type con: str or :class:`~docker.models.containers.Container`
:param command: the command to run
:type command: str
:param options: additional options to pass into Docker
:type options: kwargs
:return: ``namedtuple('ExecResult', 'exit_code,output')``
"""
if isinstance(con, str):
con = self.client.containers.get(con)
return con.exec_run(command, **options)
[docs] def service_logs(self, con):
"""Get the most recent 100 logs for ``con``
:param con: the name of the container
:type con: str
:return: a string of log messages
"""
if isinstance(con, str):
con = self.client.containers.get(con)
return con.logs(tail=100, timestamps=True).decode("utf-8")