docker-autotag/dockerautotag/Cli.py

130 lines
3.6 KiB
Python

"""Main program."""
import argparse
import copy
import os
from collections import defaultdict
import semantic_version
from dockerautotag import __version__
from dockerautotag.Logging import SingleLog
from dockerautotag.Utils import normalize_path
from dockerautotag.Utils import to_bool
from dockerautotag.Utils import to_prerelease
from dockerautotag.Utils import trim_prefix
class Autotag:
"""Handles tag operations."""
def __init__(self):
self.log = SingleLog()
self.logger = self.log.logger
self.args = self._cli_args()
self.config = self._config()
self.run()
def _cli_args(self):
parser = argparse.ArgumentParser(
description=("Creates a list of docker tags from a given version string.")
)
parser.add_argument(
"--version", action="version", version="%(prog)s {}".format(__version__)
)
return parser.parse_args()
def _config(self):
config = defaultdict(dict)
output_raw = os.environ.get("DOCKER_AUTOTAG_OUTPUT_FILE", None)
config["file"] = normalize_path(output_raw)
config["suffix"] = os.environ.get("DOCKER_AUTOTAG_SUFFIX", None)
config["version"] = os.environ.get("DOCKER_AUTOTAG_VERSION", None)
config["extra"] = os.environ.get("DOCKER_AUTOTAG_EXTRA_TAGS", None)
config["force_latest"] = to_bool(os.environ.get("DOCKER_AUTOTAG_FORCE_LATEST", False))
config["ignore_pre"] = to_bool(os.environ.get("DOCKER_AUTOTAG_IGNORE_PRERELEASE", False))
return config
@staticmethod
def _tag_extra(tags, extra):
e = []
if extra:
e = [x.strip() for x in extra.split(",")]
return tags + e
@staticmethod
def _tag_suffix(tags, suffix):
if not suffix:
return tags
res = copy.deepcopy(tags)
for t in tags:
if t == "latest":
res.append(suffix)
else:
res.append("{}-{}".format(t, suffix))
return res
@staticmethod
def _default_tags(ref, ignore_pre, force_latest):
default = ["latest"]
tags = []
if force_latest:
tags.append("latest")
if not ref:
return default
ref = trim_prefix(ref, "refs/tags/")
ref = trim_prefix(ref, "v")
try:
version = semantic_version.Version(ref)
except ValueError:
try:
version = semantic_version.Version.coerce(ref)
except Exception:
return default
except Exception:
return default
if version.prerelease:
tags.append(
"{}.{}.{}-{}".format(
version.major, version.minor, version.patch, to_prerelease(version.prerelease)
)
)
if not ignore_pre:
return tags
tags.append("{}.{}".format(version.major, version.minor))
tags.append("{}.{}.{}".format(version.major, version.minor, version.patch))
if version.major > 0:
tags.append("{}".format(version.major))
return tags
def run(self):
config = self.config
v = self._default_tags(config["version"], config["ignore_pre"], config["force_latest"])
v = self._tag_suffix(v, config["suffix"])
v = self._tag_extra(v, config["extra"])
if config["file"]:
try:
with open(config["file"], "w") as f:
f.write(",".join(v))
except IOError as e:
self.logger.error("Unable to write file: {}".format(str(e)))
print(",".join(v))