Support computed values (#21)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

changelog.d/feature.efedac04.entry.yaml

@@ -0,0 +1,5 @@
+message: Add support for computed values.
+pr_ids:
+- '21'
+timestamp: 1662456941
+type: feature

changelogd/changelogd.py

@@ -15,6 +15,7 @@
 
 from ruamel.yaml import YAML  # type: ignore
 
+from .computed_values import ComputedValueProcessor
 from .config import Config
 from .config import DEFAULT_USER_DATA
 from changelogd.resolver import Resolver
@@ -73,6 +74,9 @@ def _is_int(input: typing.Any) -> bool:
 
 def entry(config: Config, options: typing.Dict[str, typing.Optional[str]]) -> None:
     data = config.get_data()
+    computed_value_processors = [
+        ComputedValueProcessor(item) for item in data.get("computed_values", [])
+    ]
     entry_fields = [EntryField(**entry) for entry in data.get("entry_fields", [])]
     entry_type = _get_entry_type(data, options)
 
@@ -83,6 +87,10 @@ def entry(config: Config, options: typing.Dict[str, typing.Optional[str]]) -> No
 
     _add_user_data(entry, config.get_value("user_data", DEFAULT_USER_DATA))
 
+    if computed_value_processors:
+        for processor in computed_value_processors:
+            entry.update(processor.get_data())
+
     hash = hashlib.md5()
     entries_flat = " ".join(f"{key}={value}" for key, value in entry.items())
     hash.update(entries_flat.encode())

changelogd/computed_values.py

@@ -0,0 +1,88 @@
+import logging
+import re
+import subprocess
+import sys
+import typing
+from typing import List
+from typing import Optional
+
+
+def remote_branch_name() -> Optional[str]:
+    """Extract remote branch name"""
+    return _value_from_process(
+        ["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"],
+        "remote branch name",
+    )
+
+
+def local_branch_name() -> Optional[str]:
+    """Extract local branch name"""
+    return _value_from_process(
+        ["git", "rev-parse", "--abbrev-ref", "HEAD"], "local branch name"
+    )
+
+
+def branch_name() -> Optional[str]:
+    """Extract local AND remote branch name separated by space"""
+    data = []
+    local = local_branch_name()
+    if local:
+        data.append(local)
+    remote = remote_branch_name()
+    if remote:
+        data.append(remote)
+    result = " - ".join(data)
+    return result or None
+
+
+def _value_from_process(
+    command: List[str], error_context: Optional[str] = None
+) -> Optional[str]:
+    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    out, err = process.communicate()
+    if process.returncode:
+        if error_context:
+            error_context = f" to get {error_context}"
+        else:
+            error_context = ""
+        logging.error(f"Failed to run '{' '.join(command)}'{error_context}")
+        logging.error(err.decode())
+        return None
+    return out.decode()
+
+
+class ComputedValueProcessor:
+    FUNCTIONS = (local_branch_name, remote_branch_name, branch_name)
+
+    def __init__(self, data: dict):
+        type_ = data.get("type", None)
+        if not type_:
+            sys.exit(f"Missing `type` for computed value: {dict(**data)}")
+        function: typing.Optional[typing.Callable[[], Optional[str]]] = next(
+            (function for function in self.FUNCTIONS if function.__name__ == type_),
+            None,
+        )
+        if not function:
+            available_types = [function.__name__ for function in self.FUNCTIONS]
+            sys.exit(
+                f"Unavailable type: '{type_}'. "
+                f"Available types: {' '.join(available_types)}"
+            )
+        self.function: typing.Callable[[], Optional[str]] = function
+        self.name = data.get("name", None) or type_
+        self.regex = data.get("regex", None)
+        self.default = data.get("default", None)
+        self._data = data
+
+    def get_data(self) -> typing.Dict[str, typing.Any]:
+        value = self.function()
+        if self.regex:
+            match = re.search(self.regex, value) if value is not None else None
+            if match:
+                value = match.group("value")
+            else:
+                logging.warning(f"The regex '{self.regex}' didn't match '{value}'.")
+                value = None
+        if self.default and not value:
+            value = self.default
+        return {self.name: value}

docs/configuration.rst

@@ -109,3 +109,25 @@ Define fields will be captured with each entry. Available choices are:
  | - **git_email** - current user's e-mail from the git configuration.
 
 Each field's name can be changed, by defining new name after colon, e.g.: ``os_user:new_name``.
+Set the ``user_data`` value to ``null`` to avoid capturing the user data at all.
+
+computed_values
+---------------
+
+Computed values is a feature, that allows to capture a dynamic value from environment.
+The ``computed_values`` variable is a list of objects that have to define a ``type`` value.
+
+The allowed types are:
+ - ``local_branch_name`` - get the name of a local branch,
+ - ``remote_branch_name`` - get the name of a remote branch,
+ - ``branch_name`` - get the local and remote branch name separated by ``-`` (mostly 
+   suitable for running regex over it).
+
+Besides type, there are additional variables that can influence the output:
+ - ``regex`` - regular expression that will be used to extract a value from the 
+   command output. The regex need to define a named group called ``value``
+   (e.g. ``(?P<value>expression)``) which will be taken as a final value,
+ - ``name`` - name of the variable in the entry file, if not provided, the 
+   ``type`` value will be taken,
+ - ``default`` - the default value that will be used if the value (matched or
+   returned from the dynamic command) will be empty.