# docker/docker/errors.py
class DockerException(Exception):
"""Base exception class for all Docker client exceptions."""
def __init__(self, message, *args):
super(DockerException, self).__init__(message, *args)
self.explanation = None
def __str__(self):
if self.explanation:
return "%s: %s" % (super(DockerException, self).__str__(), self.explanation)
else:
return super(DockerException, self).__str__()
class APIError(DockerException):
"""
APIError is raised when the Docker Engine returns an error.
:param response: The HTTP response object.
:type response: requests.Response
:param explanation: A more descriptive explanation of the error.
:type explanation: str
:param status_code: The HTTP status code of the error.
:type status_code: int
"""
def __init__(self, response, explanation=None, status_code=None):
super(APIError, self).__init__(response.text, response.status_code)
self.response = response
self.explanation = explanation
self.status_code = status_code
def __str__(self):
# In this case, self.explanation is already formatted for the user.
return self.explanation
class NotFound(APIError):
"""
Raised when a resource is not found.
:param response: The HTTP response object.
:type response: requests.Response
:param explanation: A more descriptive explanation of the error.
:type explanation: str
"""
def __init__(self, response, explanation=None):
super(NotFound, self).__init__(response, explanation=explanation, status_code=404)
class ImageNotFound(NotFound):
"""Raised when a Docker image is not found."""
pass
class ContainerNotFound(NotFound):
"""Raised when a Docker container is not found."""
pass
class BuildError(DockerException):
"""
Raised when a docker build fails.
:param response: The HTTP response object.
:type response: requests.Response
"""
def __init__(self, response):
super(BuildError, self).__init__(response.text)
self.response = response
class InvalidVersion(DockerException):
"""
Raised when the docker version is invalid.
:param response: The HTTP response object.
:type response: requests.Response
"""
def __init__(self, response):
super(InvalidVersion, self).__init__(response.text)
self.response = response
class ServerError(DockerException):
"""
Raised when the Docker server returns an error.
:param response: The HTTP response object.
:type response: requests.Response
"""
def __init__(self, response):
super(ServerError, self).__init__(response.text)
self.response = response
class InvalidStatus(DockerException):
"""Raised when a container is in an invalid status."""
pass
class ClientError(DockerException):
"""Raised when an error occurs on the Docker client."""
pass
class TLSParameterError(DockerException):
"""Raised when TLS parameters are invalid."""
pass
class DockerError(DockerException):
"""Raised when an error occurs during the Docker API call."""
pass
class NullResource(DockerException):
"""Raised when the resource is empty."""
pass
class InvalidArgument(DockerException):
"""Raised when a command argument is invalid."""
pass
class Timeout(DockerException):
"""
Raised when a Docker operation times out.
:param message: The timeout message.
:type message: str
:param timeout: The timeout value.
:type timeout: int
"""
def __init__(self, message, timeout):
super(Timeout, self).__init__(message)
self.timeout = timeout
class ConnectionError(DockerException):
"""Raised when a connection to the Docker server fails."""
pass
class VersionError(DockerException):
"""Raised when the Docker client and server versions are incompatible."""
pass
class InvalidRepositoryName(DockerException):
"""Raised when the repository name is invalid."""
pass
# docker/docker/client.py
class Client(object):
"""
A client for the Docker Engine API.
This class provides a high-level interface to the Docker Engine API.
:param base_url: The base URL of the Docker Engine API.
:type base_url: str
:param version: The version of the Docker Engine API to use.
:type version: str
:param timeout: The timeout for API requests.
:type timeout: int
:param tls: Whether to use TLS to connect to the Docker Engine API.
:type tls: bool
:param tls_verify: Whether to verify the TLS certificate of the Docker Engine API.
:type tls_verify: bool
:param cert: The path to the TLS certificate file.
:type cert: str
:param key: The path to the TLS key file.
:type key: str
:param ca_cert: The path to the TLS CA certificate file.
:type ca_cert: str
:param max_retries: Maximum number of retries for failed requests
:type max_retries: int
:param kwargs: Additional keyword arguments passed to the ``requests.Session`` constructor.
:type kwargs: dict
"""
def __init__(self, base_url='unix://var/run/docker.sock', version='auto',
timeout=None, tls=False, tls_verify=True, cert=None, key=None,
ca_cert=None, max_retries=3, *args, **kwargs):
# ...
self.timeout = timeout
self.max_retries = max_retries
# ...
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'docker-py/%s' % __version__,
'Content-Type': 'application/json',
})
if self.tls:
self.session.cert = (cert, key)
self.session.verify = tls_verify
if ca_cert:
self.session.verify = ca_cert
# ...
self.session.max_retries = Retry(
total=max_retries,
status_forcelist=[429, 500, 502, 503, 504],
backoff_factor=0.3,
respect_retry_after_header=True,
)
# ...
