A Python library for the Docker Engine API.
—
Comprehensive exception hierarchy for robust error handling and debugging with specific exceptions for different failure scenarios.
class DockerException(Exception):
"""
Base class for all Docker SDK exceptions.
If you want to catch all errors that the Docker SDK might raise,
catch this base exception.
"""class APIError(requests.exceptions.HTTPError, DockerException):
"""
An HTTP error from the Docker API.
Attributes:
response: The HTTP response object
explanation: Detailed error explanation from Docker API
"""
class NotFound(APIError):
"""
Exception raised when a resource is not found (HTTP 404).
Raised by methods like get() when attempting to retrieve
non-existent containers, images, networks, etc.
"""
class ImageNotFound(NotFound):
"""
Exception raised when a Docker image is not found.
Specialized NotFound exception for image-related operations.
"""class InvalidVersion(DockerException):
"""
Exception raised when an invalid API version is specified.
"""
class InvalidRepository(DockerException):
"""
Exception raised when an invalid repository name is provided.
"""
class InvalidConfigFile(DockerException):
"""
Exception raised when Docker configuration file is invalid.
"""
class InvalidArgument(DockerException):
"""
Exception raised when invalid arguments are provided to API methods.
"""
class TLSParameterError(DockerException):
"""
Exception raised when TLS configuration parameters are invalid.
"""class ContainerError(DockerException):
"""
Exception raised when a container exits with a non-zero status code.
Attributes:
container: The container that failed
exit_status: The exit status code
command: The command that was executed
image: The image used
stderr: Standard error output
"""
class BuildError(DockerException):
"""
Exception raised when a Docker image build fails.
Attributes:
msg: Error message
build_log: Build log output
"""
class ImageLoadError(DockerException):
"""
Exception raised when image loading from tar archive fails.
"""
class StreamParseError(RuntimeError):
"""
Exception raised when parsing streaming API responses fails.
"""
class NullResource(DockerException, ValueError):
"""
Exception raised when attempting to perform operations on null resources.
"""
class DeprecatedMethod(DockerException):
"""
Exception raised when deprecated methods are used.
"""class MissingContextParameter(DockerException):
"""
Exception raised when required context parameters are missing.
"""
class ContextAlreadyExists(DockerException):
"""
Exception raised when attempting to create a context that already exists.
"""
class ContextException(DockerException):
"""
General exception for Docker context operations.
"""
class ContextNotFound(DockerException):
"""
Exception raised when a Docker context is not found.
"""import docker
from docker.errors import APIError, NotFound, ContainerError
client = docker.from_env()
# Handle container not found
try:
container = client.containers.get('nonexistent-container')
except NotFound:
print("Container not found")
except APIError as e:
print(f"API Error: {e.explanation}")
# Handle image not found
try:
image = client.images.get('nonexistent:image')
except docker.errors.ImageNotFound:
print("Image not found")
# Attempt to pull the image
try:
image = client.images.pull('nonexistent:image')
except APIError as e:
print(f"Failed to pull image: {e.explanation}")# Handle container execution errors
try:
container = client.containers.run(
'ubuntu:20.04',
'exit 1', # Command that will fail
remove=True
)
except ContainerError as e:
print(f"Container failed with exit code: {e.exit_status}")
print(f"Command: {e.command}")
print(f"Stderr: {e.stderr}")
# Handle container lifecycle errors
try:
container = client.containers.get('my-container')
container.start()
except NotFound:
print("Container not found")
except APIError as e:
if e.response.status_code == 409:
print("Container is already started")
else:
print(f"Failed to start container: {e.explanation}")# Handle build failures
try:
image, build_logs = client.images.build(
path='/path/to/invalid/dockerfile',
tag='my-app:latest'
)
# Process build logs
for log in build_logs:
if 'stream' in log:
print(log['stream'].strip())
except BuildError as e:
print(f"Build failed: {e.msg}")
if hasattr(e, 'build_log'):
print("Build log:")
for log_entry in e.build_log:
print(log_entry)
except APIError as e:
print(f"API error during build: {e.explanation}")# Handle network operations
try:
network = client.networks.create('my-network')
# Connect container to network
container = client.containers.get('my-container')
network.connect(container)
except APIError as e:
if 'already exists' in e.explanation.lower():
print("Network already exists")
network = client.networks.get('my-network')
else:
print(f"Network operation failed: {e.explanation}")
# Handle volume operations
try:
volume = client.volumes.create(name='my-data')
except APIError as e:
if e.response.status_code == 409:
print("Volume already exists")
volume = client.volumes.get('my-data')
else:
print(f"Volume creation failed: {e.explanation}")import docker
from docker.errors import DockerException
import logging
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_docker_operation(operation_func, *args, **kwargs):
"""
Wrapper for safe Docker operations with comprehensive error handling.
"""
try:
return operation_func(*args, **kwargs)
except ContainerError as e:
logger.error(f"Container operation failed: {e.exit_status}")
logger.error(f"Command: {e.command}")
if e.stderr:
logger.error(f"Error output: {e.stderr.decode()}")
raise
except ImageNotFound as e:
logger.error(f"Image not found: {e}")
raise
except NotFound as e:
logger.error(f"Resource not found: {e}")
raise
except BuildError as e:
logger.error(f"Build failed: {e.msg}")
raise
except TLSParameterError as e:
logger.error(f"TLS configuration error: {e}")
raise
except InvalidArgument as e:
logger.error(f"Invalid argument: {e}")
raise
except APIError as e:
logger.error(f"Docker API error ({e.response.status_code}): {e.explanation}")
raise
except DockerException as e:
logger.error(f"Docker SDK error: {e}")
raise
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise
# Usage example
client = docker.from_env()
# Safe container operations
try:
container = safe_docker_operation(
client.containers.run,
'nginx:latest',
detach=True,
name='web-server',
ports={'80/tcp': 8080}
)
logger.info(f"Container created: {container.id}")
except DockerException:
logger.error("Failed to create container")import time
import docker
from docker.errors import APIError, NotFound
def retry_operation(func, max_retries=3, delay=1, *args, **kwargs):
"""
Retry Docker operations with exponential backoff.
"""
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except APIError as e:
if attempt == max_retries - 1:
raise
if e.response.status_code >= 500: # Server errors
print(f"Attempt {attempt + 1} failed, retrying in {delay}s...")
time.sleep(delay)
delay *= 2
else:
raise # Don't retry client errors
def ensure_container_running(client, container_name, image_name):
"""
Ensure a container is running, create if necessary.
"""
try:
container = client.containers.get(container_name)
if container.status != 'running':
container.start()
return container
except NotFound:
print(f"Container {container_name} not found, creating...")
return client.containers.run(
image_name,
detach=True,
name=container_name
)
# Usage
client = docker.from_env()
try:
# Retry image pull with backoff
image = retry_operation(
client.images.pull,
max_retries=3,
repository='nginx:latest'
)
# Ensure container is running
container = ensure_container_running(
client,
'my-nginx',
'nginx:latest'
)
except DockerException as e:
print(f"Operation failed after retries: {e}")