Created
April 15, 2022 19:00
-
-
Save vergenzt/2938ca376a997cd659ae573928744718 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
''' | |
Parses current directory's git configuration + optional environment vars into a | |
list of strings suitable for tagging Docker images. | |
Outputs the computed image name(s) to stdout, separated by newlines. | |
''' | |
import os | |
import re | |
from dataclasses import dataclass | |
from functools import partial, update_wrapper | |
from itertools import product | |
from pathlib import Path | |
from shlex import split as sh_split | |
from subprocess import PIPE, run | |
from types import MappingProxyType | |
from typing import Callable, ClassVar, Dict, Generic, List, Literal, TypeVar | |
from warnings import filterwarnings, warn | |
T = TypeVar('T') | |
Getter = Callable[[], T] | |
decorate = partial | |
@dataclass | |
class WarnEmpty(Generic[T]): | |
wrapped: Getter[T] | |
on_empty: Literal['ignore', 'default', 'error'] = 'default' | |
@property | |
def _message(self): | |
return f'{self.wrapped.__name__} returned empty value' | |
def __post_init__(self): | |
update_wrapper(self, self.wrapped) | |
filterwarnings(self.on_empty, self._message) | |
def __call__(self) -> T: | |
if not (val := self.wrapped()): | |
warn(self._message, stacklevel=2) | |
return val | |
@dataclass | |
class Overrider(Generic[T]): | |
overrides: List[Getter[T]] | |
default: Getter[T] | |
def __post_init__(self): | |
update_wrapper(self, self.default) | |
def __call__(self) -> T: | |
for getter in self.overrides: | |
try: | |
return getter() | |
except NotImplementedError: | |
continue | |
return self.default() | |
@dataclass | |
class EnvVarGetter(Generic[T]): | |
env_var: str | |
env_var_value_mapper: Callable[[str], T] | |
_registry: ClassVar[Dict[str, 'EnvVarGetter']] = {} | |
@classmethod | |
@property | |
def registry(cls): | |
return MappingProxyType(cls._registry) | |
def __post_init__(self): | |
assert self.env_var not in type(self).registry | |
type(self)._registry[self.env_var] = self | |
def __call__(self) -> T: | |
if self.env_var in os.environ: | |
return self.env_var_value_mapper(os.environ[self.env_var]) | |
else: | |
raise NotImplementedError | |
@decorate(WarnEmpty) | |
@decorate(Overrider, [EnvVarGetter('DOCKER_REGISTRY_URL', str)]) | |
def docker_registry_url() -> str: | |
return '' | |
@decorate(WarnEmpty, on_empty='error') | |
@decorate(Overrider, [EnvVarGetter('DOCKER_REPO', str)]) | |
def docker_repo_name() -> str: | |
origin_url = run(sh_split('git remote get-url origin'), stdout=PIPE, text=True, check=True).stdout | |
origin_repo_name = Path(origin_url).stem | |
return origin_repo_name | |
@decorate(Overrider, [EnvVarGetter('DOCKER_TAG_SUFFIXES', str.split)]) | |
def docker_tag_suffixes() -> List[str]: | |
result = run(sh_split('git diff --quiet HEAD')) | |
try: | |
return ['-' + ({ 0: 'CLEAN', 1: 'DIRTY' }[ result.returncode ])] | |
except KeyError: | |
result.check_returncode() | |
assert False # ^ should raise | |
@decorate(Overrider, [EnvVarGetter('DOCKER_TAG_PREFIXES', str.split)]) | |
def docker_tag_prefixes() -> List[str]: | |
run(sh_split('git fetch --prune'), check=False) # try to update remotes; proceed anyway if we fail | |
refs_cmd = [ | |
'git', 'log', | |
'-1', | |
'--format=' + ' '.join([ | |
'sha/%H', # full commit SHA | |
'sha/%h' if not os.environ.get('DOCKER_TAG_OMIT_SHA_ABBREV') else '', # abbreviated commit SHA | |
'%D', # refs | |
]), | |
'--decorate=full', # prefix refs with refs/<path>/... | |
'--decorate-refs=refs/remotes/origin/*', # include (only) remote branches | |
'--decorate-refs=refs/tags/*', # include (only) tags | |
] | |
refs = run(refs_cmd, stdout=PIPE, text=True, check=True).stdout.split() | |
re_subs = [ | |
(',$', '' ), # %D separates refs with comma+space; get rid of those commas | |
('[^a-zA-Z0-9_.-]', '-' ), # sanitize chars: https://docs.docker.com/engine/reference/commandline/tag | |
('^refs-remotes-origin-', 'branch-' ), | |
('^refs-tags-', 'tag-' ), | |
] | |
for pat, repl in re_subs: | |
refs = [ re.sub(pat, repl, ref) for ref in refs ] | |
return refs | |
@decorate(WarnEmpty) | |
@decorate(Overrider, [EnvVarGetter('DOCKER_TAGS', str.split)]) | |
def docker_tags(prefixes=docker_tag_prefixes, suffixes=docker_tag_suffixes) -> List[str]: | |
return [ | |
prefix + suffix | |
for prefix in prefixes() or [''] | |
for suffix in suffixes() or [''] | |
] | |
@decorate(WarnEmpty) | |
@decorate(Overrider, [EnvVarGetter('DOCKER_IMAGE_NAMES', str.split)]) | |
def docker_image_names( | |
registry_getter: Getter[str] = docker_registry_url, | |
repo_getter: Getter[str] = docker_repo_name, | |
tags_getter: Getter[List[str]] = docker_tags | |
) -> List[str]: | |
return [ | |
''.join([ | |
'{}/'.format(registry) if registry else '', | |
repo, | |
':{}'.format(tag) if tag else '', | |
]) | |
for registry in [registry_getter() or ''] | |
for repo in [repo_getter() or ''] | |
for tag in tags_getter() or [''] | |
] | |
if __name__ == '__main__': | |
print('\n'.join(docker_image_names())) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment