mirror of
https://github.com/thegeeklab/docker-tidy.git
synced 2024-11-24 21:20:42 +00:00
fix flake8 errors
This commit is contained in:
parent
964e25d46a
commit
b7be2b4b95
2
.flake8
2
.flake8
@ -1,5 +1,5 @@
|
||||
[flake8]
|
||||
ignore = D103, W503
|
||||
ignore = D103, D107, W503
|
||||
max-line-length = 99
|
||||
inline-quotes = double
|
||||
exclude =
|
||||
|
||||
@ -12,6 +12,7 @@ from dockertidy.Parser import timedelta
|
||||
|
||||
|
||||
class AutoStop:
|
||||
"""Autostop object to handle long running containers."""
|
||||
|
||||
def __init__(self):
|
||||
self.config = SingleConfig()
|
||||
@ -20,6 +21,7 @@ class AutoStop:
|
||||
self.docker = self._get_docker_client()
|
||||
|
||||
def stop_containers(self):
|
||||
"""Identify long running containers and terminate them."""
|
||||
client = self.docker
|
||||
config = self.config.config
|
||||
|
||||
|
||||
@ -11,6 +11,7 @@ from dockertidy.Parser import timedelta_validator
|
||||
|
||||
|
||||
class DockerTidy:
|
||||
"""Cli entrypoint to handle command arguments."""
|
||||
|
||||
def __init__(self):
|
||||
self.log = SingleLog()
|
||||
|
||||
@ -20,14 +20,12 @@ default_config_file = os.path.join(config_dir, "config.yml")
|
||||
|
||||
|
||||
class Config():
|
||||
"""
|
||||
Create an object with all necessary settings.
|
||||
"""Create an object with all necessary settings.
|
||||
|
||||
Settings are loade from multiple locations in defined order (last wins):
|
||||
- default settings defined by `self._get_defaults()`
|
||||
- yaml config file, defaults to OS specific user config dir
|
||||
see (https://pypi.org/project/appdirs/)
|
||||
- provides cli parameters
|
||||
- default settings defined by `self._get_defaults()`
|
||||
- yaml config file, defaults to OS specific user config directory
|
||||
- provided cli parameters
|
||||
"""
|
||||
|
||||
SETTINGS = {
|
||||
@ -241,4 +239,6 @@ class Config():
|
||||
|
||||
|
||||
class SingleConfig(Config, metaclass=Singleton):
|
||||
"""Singleton config object."""
|
||||
|
||||
pass
|
||||
|
||||
@ -2,305 +2,309 @@
|
||||
"""Remove unused docker containers and images."""
|
||||
|
||||
import fnmatch
|
||||
import logging
|
||||
import sys
|
||||
from collections import namedtuple
|
||||
|
||||
import dateutil.parser
|
||||
import docker
|
||||
import docker.errors
|
||||
import requests.exceptions
|
||||
from docker.utils import kwargs_from_env
|
||||
|
||||
from dockertidy.Config import SingleConfig
|
||||
from dockertidy.Logger import SingleLog
|
||||
|
||||
# This seems to be something docker uses for a null/zero date
|
||||
YEAR_ZERO = "0001-01-01T00:00:00Z"
|
||||
|
||||
ExcludeLabel = namedtuple("ExcludeLabel", ["key", "value"])
|
||||
|
||||
|
||||
class GarbageCollector:
|
||||
"""Garbage collector object to handle cleanup tasks of container, images and volumes."""
|
||||
|
||||
# This seems to be something docker uses for a null/zero date
|
||||
YEAR_ZERO = "0001-01-01T00:00:00Z"
|
||||
ExcludeLabel = namedtuple("ExcludeLabel", ["key", "value"])
|
||||
|
||||
def __init__(self):
|
||||
self.config = SingleConfig()
|
||||
self.log = SingleLog()
|
||||
self.logger = SingleLog().logger
|
||||
self.docker = self._get_docker_client()
|
||||
|
||||
|
||||
def cleanup_containers(client, max_container_age, dry_run, exclude_container_labels):
|
||||
all_containers = get_all_containers(client)
|
||||
filtered_containers = filter_excluded_containers(
|
||||
all_containers,
|
||||
exclude_container_labels,
|
||||
)
|
||||
for container_summary in reversed(list(filtered_containers)):
|
||||
container = api_call(
|
||||
client.inspect_container,
|
||||
container=container_summary["Id"],
|
||||
def cleanup_containers(self):
|
||||
"""Identify old containers and remove them."""
|
||||
config = self.config.config
|
||||
client = self.docker
|
||||
all_containers = self._get_all_containers(client)
|
||||
filtered_containers = self._filter_excluded_containers(
|
||||
all_containers,
|
||||
config["gc"]["exclude_container_labels"],
|
||||
)
|
||||
if not container or not should_remove_container(
|
||||
|
||||
for container_summary in reversed(list(filtered_containers)):
|
||||
container = self._api_call(
|
||||
client.inspect_container,
|
||||
container=container_summary["Id"],
|
||||
)
|
||||
if not container or not self._should_remove_container(
|
||||
container,
|
||||
max_container_age,
|
||||
):
|
||||
continue
|
||||
config["gc"]["max_container_age"],
|
||||
):
|
||||
continue
|
||||
|
||||
log.info("Removing container %s %s %s" % (container["Id"][:16], container.get(
|
||||
"Name", "").lstrip("/"), container["State"]["FinishedAt"]))
|
||||
|
||||
if not dry_run:
|
||||
api_call(
|
||||
client.remove_container,
|
||||
container=container["Id"],
|
||||
v=True,
|
||||
self.logger.info(
|
||||
"Removing container %s %s %s" % (
|
||||
container["Id"][:16], container.get("Name", "").lstrip("/"),
|
||||
container["State"]["FinishedAt"]
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def filter_excluded_containers(containers, exclude_container_labels):
|
||||
if not exclude_container_labels:
|
||||
return containers
|
||||
|
||||
def include_container(container):
|
||||
if should_exclude_container_with_labels(
|
||||
container,
|
||||
exclude_container_labels,
|
||||
):
|
||||
return False
|
||||
return True
|
||||
|
||||
return filter(include_container, containers)
|
||||
|
||||
|
||||
def should_exclude_container_with_labels(container, exclude_container_labels):
|
||||
if container["Labels"]:
|
||||
for exclude_label in exclude_container_labels:
|
||||
if exclude_label.value:
|
||||
matching_keys = fnmatch.filter(
|
||||
container["Labels"].keys(),
|
||||
exclude_label.key,
|
||||
if not config["dry_run"]:
|
||||
self._api_call(
|
||||
client.remove_container,
|
||||
container=container["Id"],
|
||||
v=True,
|
||||
)
|
||||
label_values_to_check = [
|
||||
container["Labels"][matching_key] for matching_key in matching_keys
|
||||
]
|
||||
if fnmatch.filter(label_values_to_check, exclude_label.value):
|
||||
return True
|
||||
else:
|
||||
if fnmatch.filter(container["Labels"].keys(), exclude_label.key):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _filter_excluded_containers(self, containers):
|
||||
config = self.config.config
|
||||
|
||||
def should_remove_container(container, min_date):
|
||||
state = container.get("State", {})
|
||||
if not config["gc"]["exclude_container_labels"]:
|
||||
return containers
|
||||
|
||||
if state.get("Running"):
|
||||
def include_container(container):
|
||||
if self._should_exclude_container_with_labels(
|
||||
container,
|
||||
config["gc"]["exclude_container_labels"],
|
||||
):
|
||||
return False
|
||||
return True
|
||||
|
||||
return filter(include_container, containers)
|
||||
|
||||
def _should_exclude_container_with_labels(self, container):
|
||||
config = self.config.config
|
||||
|
||||
if container["Labels"]:
|
||||
for exclude_label in config["gc"]["exclude_container_labels"]:
|
||||
if exclude_label.value:
|
||||
matching_keys = fnmatch.filter(
|
||||
container["Labels"].keys(),
|
||||
exclude_label.key,
|
||||
)
|
||||
label_values_to_check = [
|
||||
container["Labels"][matching_key] for matching_key in matching_keys
|
||||
]
|
||||
if fnmatch.filter(label_values_to_check, exclude_label.value):
|
||||
return True
|
||||
else:
|
||||
if fnmatch.filter(container["Labels"].keys(), exclude_label.key):
|
||||
return True
|
||||
return False
|
||||
|
||||
if state.get("Ghost"):
|
||||
return True
|
||||
def _should_remove_container(self, container, min_date):
|
||||
state = container.get("State", {})
|
||||
|
||||
# Container was created, but never started
|
||||
if state.get("FinishedAt") == YEAR_ZERO:
|
||||
created_date = dateutil.parser.parse(container["Created"])
|
||||
return created_date < min_date
|
||||
if state.get("Running"):
|
||||
return False
|
||||
|
||||
finished_date = dateutil.parser.parse(state["FinishedAt"])
|
||||
return finished_date < min_date
|
||||
|
||||
|
||||
def get_all_containers(client):
|
||||
log.info("Getting all containers")
|
||||
containers = client.containers(all=True)
|
||||
log.info("Found %s containers", len(containers))
|
||||
return containers
|
||||
|
||||
|
||||
def get_all_images(client):
|
||||
log.info("Getting all images")
|
||||
images = client.images()
|
||||
log.info("Found %s images", len(images))
|
||||
return images
|
||||
|
||||
|
||||
def get_dangling_volumes(client):
|
||||
log.info("Getting dangling volumes")
|
||||
volumes = client.volumes({"dangling": True})["Volumes"] or []
|
||||
log.info("Found %s dangling volumes", len(volumes))
|
||||
return volumes
|
||||
|
||||
|
||||
def cleanup_images(client, max_image_age, dry_run, exclude_set):
|
||||
# re-fetch container list so that we don't include removed containers
|
||||
|
||||
containers = get_all_containers(client)
|
||||
images = get_all_images(client)
|
||||
if docker.utils.compare_version("1.21", client._version) < 0:
|
||||
image_tags_in_use = {container["Image"] for container in containers}
|
||||
images = filter_images_in_use(images, image_tags_in_use)
|
||||
else:
|
||||
# ImageID field was added in 1.21
|
||||
image_ids_in_use = {container["ImageID"] for container in containers}
|
||||
images = filter_images_in_use_by_id(images, image_ids_in_use)
|
||||
images = filter_excluded_images(images, exclude_set)
|
||||
|
||||
for image_summary in reversed(list(images)):
|
||||
remove_image(client, image_summary, max_image_age, dry_run)
|
||||
|
||||
|
||||
def filter_excluded_images(images, exclude_set):
|
||||
|
||||
def include_image(image_summary):
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
if no_image_tags(image_tags):
|
||||
if state.get("Ghost"):
|
||||
return True
|
||||
for exclude_pattern in exclude_set:
|
||||
if fnmatch.filter(image_tags, exclude_pattern):
|
||||
return False
|
||||
return True
|
||||
|
||||
return filter(include_image, images)
|
||||
# Container was created, but never started
|
||||
if state.get("FinishedAt") == self.YEAR_ZERO:
|
||||
created_date = dateutil.parser.parse(container["Created"])
|
||||
return created_date < min_date
|
||||
|
||||
finished_date = dateutil.parser.parse(state["FinishedAt"])
|
||||
return finished_date < min_date
|
||||
|
||||
def filter_images_in_use(images, image_tags_in_use):
|
||||
def _get_all_containers(self):
|
||||
client = self.docker
|
||||
self.logger.info("Getting all containers")
|
||||
containers = client.containers(all=True)
|
||||
self.logger.info("Found %s containers", len(containers))
|
||||
return containers
|
||||
|
||||
def get_tag_set(image_summary):
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
if no_image_tags(image_tags):
|
||||
# The repr of the image Id used by client.containers()
|
||||
return set(["%s:latest" % image_summary["Id"][:12]])
|
||||
return set(image_tags)
|
||||
def _get_all_images(self):
|
||||
client = self.docker
|
||||
self.logger.info("Getting all images")
|
||||
images = client.images()
|
||||
self.logger.info("Found %s images", len(images))
|
||||
return images
|
||||
|
||||
def image_not_in_use(image_summary):
|
||||
return not get_tag_set(image_summary) & image_tags_in_use
|
||||
def _get_dangling_volumes(self):
|
||||
client = self.docker
|
||||
self.logger.info("Getting dangling volumes")
|
||||
volumes = client.volumes({"dangling": True})["Volumes"] or []
|
||||
self.logger.info("Found %s dangling volumes", len(volumes))
|
||||
return volumes
|
||||
|
||||
return filter(image_not_in_use, images)
|
||||
def cleanup_images(self, exclude_set):
|
||||
"""Identify old images and remove them."""
|
||||
# re-fetch container list so that we don't include removed containers
|
||||
client = self.docker
|
||||
config = self.config.config
|
||||
|
||||
|
||||
def filter_images_in_use_by_id(images, image_ids_in_use):
|
||||
|
||||
def image_not_in_use(image_summary):
|
||||
return image_summary["Id"] not in image_ids_in_use
|
||||
|
||||
return filter(image_not_in_use, images)
|
||||
|
||||
|
||||
def is_image_old(image, min_date):
|
||||
return dateutil.parser.parse(image["Created"]) < min_date
|
||||
|
||||
|
||||
def no_image_tags(image_tags):
|
||||
return not image_tags or image_tags == ["<none>:<none>"]
|
||||
|
||||
|
||||
def remove_image(client, image_summary, min_date, dry_run):
|
||||
image = api_call(client.inspect_image, image=image_summary["Id"])
|
||||
if not image or not is_image_old(image, min_date):
|
||||
return
|
||||
|
||||
log.info("Removing image %s" % format_image(image, image_summary))
|
||||
if dry_run:
|
||||
return
|
||||
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
# If there are no tags, remove the id
|
||||
if no_image_tags(image_tags):
|
||||
api_call(client.remove_image, image=image_summary["Id"])
|
||||
return
|
||||
|
||||
# Remove any repository tags so we don't hit 409 Conflict
|
||||
for image_tag in image_tags:
|
||||
api_call(client.remove_image, image=image_tag)
|
||||
|
||||
|
||||
def remove_volume(client, volume, dry_run):
|
||||
if not volume:
|
||||
return
|
||||
|
||||
log.info("Removing volume %s" % volume["Name"])
|
||||
if dry_run:
|
||||
return
|
||||
|
||||
api_call(client.remove_volume, name=volume["Name"])
|
||||
|
||||
|
||||
def cleanup_volumes(client, dry_run):
|
||||
dangling_volumes = get_dangling_volumes(client)
|
||||
|
||||
for volume in reversed(dangling_volumes):
|
||||
log.info("Removing dangling volume %s", volume["Name"])
|
||||
remove_volume(client, volume, dry_run)
|
||||
|
||||
|
||||
def api_call(func, **kwargs):
|
||||
try:
|
||||
return func(**kwargs)
|
||||
except requests.exceptions.Timeout as e:
|
||||
params = ",".join("%s=%s" % item for item in kwargs.items())
|
||||
log.warn("Failed to call %s %s %s" % (func.__name__, params, e))
|
||||
except docker.errors.APIError as ae:
|
||||
params = ",".join("%s=%s" % item for item in kwargs.items())
|
||||
log.warn("Error calling %s %s %s" % (func.__name__, params, ae))
|
||||
|
||||
|
||||
def format_image(image, image_summary):
|
||||
|
||||
def get_tags():
|
||||
tags = image_summary.get("RepoTags")
|
||||
if not tags or tags == ["<none>:<none>"]:
|
||||
return ""
|
||||
return ", ".join(tags)
|
||||
|
||||
return "%s %s" % (image["Id"][:16], get_tags())
|
||||
|
||||
|
||||
def build_exclude_set(image_tags, exclude_file):
|
||||
exclude_set = set(image_tags or [])
|
||||
|
||||
def is_image_tag(line):
|
||||
return line and not line.startswith("#")
|
||||
|
||||
if exclude_file:
|
||||
lines = [line.strip() for line in exclude_file.read().split("\n")]
|
||||
exclude_set.update(filter(is_image_tag, lines))
|
||||
return exclude_set
|
||||
|
||||
|
||||
def format_exclude_labels(exclude_label_args):
|
||||
exclude_labels = []
|
||||
for exclude_label_arg in exclude_label_args:
|
||||
split_exclude_label = exclude_label_arg.split("=", 1)
|
||||
exclude_label_key = split_exclude_label[0]
|
||||
if len(split_exclude_label) == 2:
|
||||
exclude_label_value = split_exclude_label[1]
|
||||
containers = self._get_all_containers()
|
||||
images = self._get_all_images()
|
||||
if docker.utils.compare_version("1.21", client._version) < 0:
|
||||
image_tags_in_use = {container["Image"] for container in containers}
|
||||
images = self._filter_images_in_use(images, image_tags_in_use)
|
||||
else:
|
||||
exclude_label_value = None
|
||||
exclude_labels.append(ExcludeLabel(
|
||||
key=exclude_label_key,
|
||||
value=exclude_label_value,
|
||||
))
|
||||
return exclude_labels
|
||||
# ImageID field was added in 1.21
|
||||
image_ids_in_use = {container["ImageID"] for container in containers}
|
||||
images = self._filter_images_in_use_by_id(images, image_ids_in_use)
|
||||
images = self._filter_excluded_images(images, exclude_set)
|
||||
|
||||
for image_summary in reversed(list(images)):
|
||||
self._rmove_image(
|
||||
client, image_summary, config["gc"]["max_image_age"], config["dry_run"]
|
||||
)
|
||||
|
||||
def _filter_excluded_images(self, images, exclude_set):
|
||||
|
||||
def include_image(image_summary):
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
if self.no_image_tags(image_tags):
|
||||
return True
|
||||
for exclude_pattern in exclude_set:
|
||||
if fnmatch.filter(image_tags, exclude_pattern):
|
||||
return False
|
||||
return True
|
||||
|
||||
return filter(include_image, images)
|
||||
|
||||
def _filter_images_in_use(self, images, image_tags_in_use):
|
||||
|
||||
def get_tag_set(image_summary):
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
if self._no_image_tags(image_tags):
|
||||
# The repr of the image Id used by client.containers()
|
||||
return set(["%s:latest" % image_summary["Id"][:12]])
|
||||
return set(image_tags)
|
||||
|
||||
def image_not_in_use(image_summary):
|
||||
return not get_tag_set(image_summary) & image_tags_in_use
|
||||
|
||||
return filter(image_not_in_use, images)
|
||||
|
||||
def _filter_images_in_use_by_id(self, images, image_ids_in_use):
|
||||
|
||||
def image_not_in_use(image_summary):
|
||||
return image_summary["Id"] not in image_ids_in_use
|
||||
|
||||
return filter(image_not_in_use, images)
|
||||
|
||||
def _is_image_old(self, image, min_date):
|
||||
return dateutil.parser.parse(image["Created"]) < min_date
|
||||
|
||||
def _no_image_tags(self, image_tags):
|
||||
return not image_tags or image_tags == ["<none>:<none>"]
|
||||
|
||||
def _remove_image(self, client, image_summary, min_date):
|
||||
config = self.config.config
|
||||
client = self.docker
|
||||
image = self._api_call(client.inspect_image, image=image_summary["Id"])
|
||||
|
||||
if not image or not self._is_image_old(image, min_date):
|
||||
return
|
||||
|
||||
self.logger.info("Removing image %s" % self._format_image(image, image_summary))
|
||||
if config["dry_run"]:
|
||||
return
|
||||
|
||||
image_tags = image_summary.get("RepoTags")
|
||||
# If there are no tags, remove the id
|
||||
if self._no_image_tags(image_tags):
|
||||
self._api_call(client.remove_image, image=image_summary["Id"])
|
||||
return
|
||||
|
||||
# Remove any repository tags so we don't hit 409 Conflict
|
||||
for image_tag in image_tags:
|
||||
self._api_call(client.remove_image, image=image_tag)
|
||||
|
||||
def _remove_volume(self, client, volume):
|
||||
config = self.config.config
|
||||
if not volume:
|
||||
return
|
||||
|
||||
self.logger.info("Removing volume %s" % volume["Name"])
|
||||
if config["dry_run"]:
|
||||
return
|
||||
|
||||
self._api_call(client.remove_volume, name=volume["Name"])
|
||||
|
||||
def cleanup_volumes(self, client):
|
||||
"""Identify old volumes and remove them."""
|
||||
dangling_volumes = self._get_dangling_volumes(client)
|
||||
|
||||
for volume in reversed(dangling_volumes):
|
||||
self.logger.info("Removing dangling volume %s", volume["Name"])
|
||||
self._remove_volume(client, volume)
|
||||
|
||||
def _api_call(self, func, **kwargs):
|
||||
try:
|
||||
return func(**kwargs)
|
||||
except requests.exceptions.Timeout as e:
|
||||
params = ",".join("%s=%s" % item for item in kwargs.items())
|
||||
self.logger.warn("Failed to call %s %s %s" % (func.__name__, params, e))
|
||||
except docker.errors.APIError as ae:
|
||||
params = ",".join("%s=%s" % item for item in kwargs.items())
|
||||
self.logger.warn("Error calling %s %s %s" % (func.__name__, params, ae))
|
||||
|
||||
def _format_image(self, image, image_summary):
|
||||
|
||||
def get_tags():
|
||||
tags = image_summary.get("RepoTags")
|
||||
if not tags or tags == ["<none>:<none>"]:
|
||||
return ""
|
||||
return ", ".join(tags)
|
||||
|
||||
return "%s %s" % (image["Id"][:16], get_tags())
|
||||
|
||||
def _build_exclude_set(self):
|
||||
config = self.config.config
|
||||
exclude_set = set(config["gc"]["exclude_image"])
|
||||
|
||||
def is_image_tag(line):
|
||||
return line and not line.startswith("#")
|
||||
|
||||
return exclude_set
|
||||
|
||||
def _format_exclude_labels(self):
|
||||
config = self.config.config
|
||||
exclude_labels = []
|
||||
|
||||
for exclude_label_arg in config["gc"]["exclude_container_label"]:
|
||||
split_exclude_label = exclude_label_arg.split("=", 1)
|
||||
exclude_label_key = split_exclude_label[0]
|
||||
if len(split_exclude_label) == 2:
|
||||
exclude_label_value = split_exclude_label[1]
|
||||
else:
|
||||
exclude_label_value = None
|
||||
exclude_labels.append(
|
||||
self.ExcludeLabel(
|
||||
key=exclude_label_key,
|
||||
value=exclude_label_value,
|
||||
)
|
||||
)
|
||||
return exclude_labels
|
||||
|
||||
def _get_docker_client(self):
|
||||
config = self.config.config
|
||||
return docker.APIClient(version="auto", timeout=config["timeout"])
|
||||
|
||||
|
||||
def main():
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout)
|
||||
# def main():
|
||||
# exclude_container_labels = format_exclude_labels(args.exclude_container_label)
|
||||
|
||||
args = get_args()
|
||||
client = docker.APIClient(version="auto", timeout=args.timeout, **kwargs_from_env())
|
||||
# if args.max_container_age:
|
||||
# cleanup_containers(
|
||||
# client,
|
||||
# args.max_container_age,
|
||||
# args.dry_run,
|
||||
# exclude_container_labels,
|
||||
# )
|
||||
|
||||
exclude_container_labels = format_exclude_labels(args.exclude_container_label)
|
||||
# if args.max_image_age:
|
||||
# exclude_set = build_exclude_set(args.exclude_image, args.exclude_image_file)
|
||||
# cleanup_images(client, args.max_image_age, args.dry_run, exclude_set)
|
||||
|
||||
if args.max_container_age:
|
||||
cleanup_containers(
|
||||
client,
|
||||
args.max_container_age,
|
||||
args.dry_run,
|
||||
exclude_container_labels,
|
||||
)
|
||||
|
||||
if args.max_image_age:
|
||||
exclude_set = build_exclude_set(args.exclude_image, args.exclude_image_file)
|
||||
cleanup_images(client, args.max_image_age, args.dry_run, exclude_set)
|
||||
|
||||
if args.dangling_volumes:
|
||||
cleanup_volumes(client, args.dry_run)
|
||||
# if args.dangling_volumes:
|
||||
# cleanup_volumes(client, args.dry_run)
|
||||
|
||||
@ -61,6 +61,7 @@ class MultilineJsonFormatter(jsonlogger.JsonFormatter):
|
||||
|
||||
|
||||
class Log:
|
||||
"""Base logging object."""
|
||||
|
||||
def __init__(self, level=logging.WARN, name="dockertidy", json=False):
|
||||
self.logger = logging.getLogger(name)
|
||||
@ -78,7 +79,9 @@ class Log:
|
||||
handler.addFilter(LogFilter(logging.ERROR))
|
||||
handler.setFormatter(
|
||||
MultilineFormatter(
|
||||
self.error(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))))
|
||||
self.error(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))
|
||||
)
|
||||
)
|
||||
|
||||
if json:
|
||||
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
||||
@ -91,7 +94,9 @@ class Log:
|
||||
handler.addFilter(LogFilter(logging.WARN))
|
||||
handler.setFormatter(
|
||||
MultilineFormatter(
|
||||
self.warn(CONSOLE_FORMAT.format(colorama.Fore.YELLOW, colorama.Style.RESET_ALL))))
|
||||
self.warn(CONSOLE_FORMAT.format(colorama.Fore.YELLOW, colorama.Style.RESET_ALL))
|
||||
)
|
||||
)
|
||||
|
||||
if json:
|
||||
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
||||
@ -104,7 +109,9 @@ class Log:
|
||||
handler.addFilter(LogFilter(logging.INFO))
|
||||
handler.setFormatter(
|
||||
MultilineFormatter(
|
||||
self.info(CONSOLE_FORMAT.format(colorama.Fore.CYAN, colorama.Style.RESET_ALL))))
|
||||
self.info(CONSOLE_FORMAT.format(colorama.Fore.CYAN, colorama.Style.RESET_ALL))
|
||||
)
|
||||
)
|
||||
|
||||
if json:
|
||||
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
||||
@ -117,7 +124,9 @@ class Log:
|
||||
handler.addFilter(LogFilter(logging.CRITICAL))
|
||||
handler.setFormatter(
|
||||
MultilineFormatter(
|
||||
self.critical(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))))
|
||||
self.critical(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))
|
||||
)
|
||||
)
|
||||
|
||||
if json:
|
||||
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
||||
@ -130,8 +139,9 @@ class Log:
|
||||
handler.addFilter(LogFilter(logging.DEBUG))
|
||||
handler.setFormatter(
|
||||
MultilineFormatter(
|
||||
self.critical(CONSOLE_FORMAT.format(colorama.Fore.BLUE,
|
||||
colorama.Style.RESET_ALL))))
|
||||
self.critical(CONSOLE_FORMAT.format(colorama.Fore.BLUE, colorama.Style.RESET_ALL))
|
||||
)
|
||||
)
|
||||
|
||||
if json:
|
||||
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
||||
@ -139,6 +149,7 @@ class Log:
|
||||
return handler
|
||||
|
||||
def set_level(self, s):
|
||||
"""Set log level."""
|
||||
self.logger.setLevel(s)
|
||||
|
||||
def debug(self, msg):
|
||||
@ -173,12 +184,16 @@ class Log:
|
||||
return "{}{}{}".format(color, msg, colorama.Style.RESET_ALL)
|
||||
|
||||
def sysexit(self, code=1):
|
||||
"""Exit running program with given exit code."""
|
||||
sys.exit(code)
|
||||
|
||||
def sysexit_with_message(self, msg, code=1):
|
||||
"""Exit running program with given exit code and message."""
|
||||
self.logger.critical(str(msg))
|
||||
self.sysexit(code)
|
||||
|
||||
|
||||
class SingleLog(Log, metaclass=Singleton):
|
||||
"""Singleton logger object."""
|
||||
|
||||
pass
|
||||
|
||||
@ -9,9 +9,12 @@ def to_bool(string):
|
||||
|
||||
|
||||
class Singleton(type):
|
||||
"""Singleton metaclass."""
|
||||
|
||||
_instances = {}
|
||||
|
||||
def __call__(cls, *args, **kwargs):
|
||||
def __call__(cls, *args, **kwargs): # noqa
|
||||
|
||||
if cls not in cls._instances:
|
||||
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
|
||||
return cls._instances[cls]
|
||||
|
||||
Loading…
Reference in New Issue
Block a user