docker_image_names.py

#!/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()))