feat: Update post process to generate link using a link parser, and b…

…uild body with a jinja template. (#48)

closes #44

changelog_gen/cli/command.py

@@ -321,8 +321,7 @@ def _gen(  # noqa: PLR0913, C901, PLR0915
 
     w = writer.new_writer(context, extension, dry_run=dry_run, change_template=cfg.change_template)
 
-    w.add_version(version_string)
-    w.consume(cfg.type_headers, changes)
+    w.consume(version_string, cfg.type_headers, changes)
 
     change_lines = create_with_editor(context, str(w), extension) if interactive else str(w)
 
@@ -382,5 +381,4 @@ def release_hook(_context: Context, new_version: str) -> list[str]:
             context.error("httpx required to execute post process, install with `--extras post-process`.")
             return
 
-        unique_issues = [r.issue_ref.replace("#", "") for r in changes if r.issue_ref]
-        per_issue_post_process(context, post_process, sorted(unique_issues), str(new), dry_run=dry_run)
+        per_issue_post_process(context, post_process, changes, str(new), dry_run=dry_run)

changelog_gen/config.py

@@ -37,11 +37,11 @@
 class PostProcessConfig:
     """Post Processor configuration options."""
 
-    url: str | None = None
+    link_parser: dict[str, str] | None = None
     verb: str = "POST"
     # The body to send as a post-processing command,
     # can have the entries: ::issue_ref::, ::version::
-    body: str = '{"body": "Released on ::version::"}'
+    body_template: str = '{"body": "Released on {{ version }}"}'
     auth_type: str = "basic"  # future proof config
     headers: dict | None = None
     # Name of an environment variable to use as HTTP Basic Auth parameters.
@@ -154,24 +154,12 @@ def to_dict(self: Config) -> dict:
 
 
 @timer
-def _process_overrides(overrides: dict) -> tuple[dict, PostProcessConfig | None]:
+def _process_overrides(overrides: dict) -> dict:
     """Process provided overrides.
 
     Remove any unsupplied values (None).
     """
-    post_process_url = overrides.pop("post_process_url", "")
-    post_process_auth_env = overrides.pop("post_process_auth_env", None)
-
-    post_process = None
-    if post_process_url or post_process_auth_env:
-        post_process = PostProcessConfig(
-            url=post_process_url,
-            auth_env=post_process_auth_env,
-        )
-
-    overrides = {k: v for k, v in overrides.items() if v is not None}
-
-    return overrides, post_process
+    return {k: v for k, v in overrides.items() if v is not None}
 
 
 @timer
@@ -205,13 +193,13 @@ def check_deprecations(cfg: dict) -> None:  # noqa: ARG001
 
 
 @timer
-def read(path: str = "pyproject.toml", **kwargs) -> Config:  # noqa: C901
+def read(path: str = "pyproject.toml", **kwargs) -> Config:
     """Read configuration from local environment.
 
     Supported configuration locations (checked in order):
     * pyproject.toml
     """
-    overrides, post_process = _process_overrides(kwargs)
+    overrides = _process_overrides(kwargs)
     cfg = {}
 
     pyproject = Path(path)
@@ -223,37 +211,10 @@ def read(path: str = "pyproject.toml", **kwargs) -> Config:  # noqa: C901
     # parse pyproject
     cfg = _process_pyproject(pyproject)
 
-    if "post_process" not in cfg and post_process:
-        cfg["post_process"] = {
-            "url": post_process.url,
-            "auth_env": post_process.auth_env,
-        }
-
-    if "post_process" in cfg and post_process:
-        cfg["post_process"]["url"] = post_process.url or cfg["post_process"].get("url")
-        cfg["post_process"]["auth_env"] = post_process.auth_env or cfg["post_process"].get("auth_env")
-
     cfg.update(overrides)
 
     check_deprecations(cfg)  # pragma: no mutate
 
-    for replace_key_path in [
-        ("post_process", "url"),
-        ("post_process", "body"),
-    ]:
-        data, value = cfg, None
-        for key in replace_key_path:
-            value = data.get(key)
-            if key in data:
-                data = data[key]
-
-        # check for non supported replace keys
-        supported = {"::issue_ref::", "::version::", "::commit_hash::", "::pull_ref::"}
-        unsupported = sorted(set(re.findall(r"(::.*?::)", str(value)) or []) - supported)
-        if unsupported:
-            msg = f"""Replace string(s) ('{"', '".join(unsupported)}') not supported."""
-            raise errors.UnsupportedReplaceError(msg)
-
     if cfg.get("post_process"):
         pp = cfg["post_process"]
         try:

changelog_gen/extractor.py

@@ -36,6 +36,7 @@ class Change:  # noqa: D101
     breaking: bool = False
     footers: list[Footer] = dataclasses.field(default_factory=list)
     links: list[Link] = dataclasses.field(default_factory=list)
+    rendered: str = ""  # This is populated by the writer at run time
 
     def __lt__(self: t.Self, other: Change) -> bool:  # noqa: D105
         s = (not self.breaking, self.scope.lower() if self.scope else "zzz", self.issue_ref.lower())

changelog_gen/post_processor.py

@@ -1,17 +1,21 @@
 from __future__ import annotations
 
 import os
+import re
 import typing as t
 from http import HTTPStatus
 
 import httpx
 import typer
+from jinja2 import BaseLoader, Environment
 
+from changelog_gen.extractor import Link
 from changelog_gen.util import timer
 
 if t.TYPE_CHECKING:
     from changelog_gen.config import PostProcessConfig
     from changelog_gen.context import Context
+    from changelog_gen.extractor import Change
 
 
 class BearerAuth(httpx.Auth):
@@ -56,39 +60,53 @@ def make_client(context: Context, cfg: PostProcessConfig) -> httpx.Client:
     )
 
 
+def _get_footer(change: Change, footer: str) -> str | None:
+    for footer_ in change.footers:
+        if footer_.footer.lower() == footer.lower():
+            return footer_
+    return None
+
+
 @timer
 def per_issue_post_process(
     context: Context,
     cfg: PostProcessConfig,
-    issue_refs: list[str],
+    changes: list[Change],
     version_tag: str,
     *,
     dry_run: bool = False,
 ) -> None:
     """Run post process for all provided issue references."""
-    if not cfg.url:
+    link_parser, body_template = cfg.link_parser, cfg.body_template
+    if link_parser is None:
         return
     context.warning("Post processing:")
 
     client = make_client(context, cfg)
 
-    for issue in issue_refs:
-        url, body = cfg.url, cfg.body
-        for find, replace in [
-            ("::issue_ref::", issue),
-            ("::version::", version_tag),
-        ]:
-            url = url.replace(find, replace)
-            body = body.replace(find, replace)
+    for change in changes:
+        link = None
+
+        footer = _get_footer(change, link_parser["target"])
+        if footer is None:
+            continue
+
+        text_template = link_parser.get("text", "{0}")
+        link_template = link_parser["link"]
+        matches = re.findall(link_parser["pattern"], footer.value)
+        link = Link(text_template.format(matches[0]), link_template.format(matches[0]))
+
+        rtemplate = Environment(loader=BaseLoader()).from_string(body_template)  # noqa: S701
 
+        body = rtemplate.render(link=link, change=change, version=version_tag)
         context.indent()
         if dry_run:
-            context.warning("Would request: %s %s %s", cfg.verb, url, body)
+            context.warning("Would request: %s %s %s", cfg.verb, link.link, body)
         else:
-            context.info("Request: %s %s", cfg.verb, url)
+            context.info("Request: %s %s", cfg.verb, link.link)
             r = client.request(
                 method=cfg.verb,
-                url=url,
+                url=link.link,
                 content=body,
             )
             context.indent()

changelog_gen/writer.py

@@ -51,48 +51,33 @@ def __init__(
         self._change_template = change_template
 
     @timer
-    def add_version(self: t.Self, version: str) -> None:
-        """Add a version string to changelog file."""
-        self._add_version(version)
+    def _render_change(self: t.Self, change: Change) -> str:
+        rtemplate = Environment(loader=BaseLoader()).from_string(self._change_template.replace("\n", ""))  # noqa: S701
 
-    @timer
-    def _add_version(self: t.Self, version: str) -> None:
-        raise NotImplementedError
+        return rtemplate.render(change=change)
 
     @timer
-    def consume(self: t.Self, type_headers: dict[str, str], changes: list[Change]) -> None:
+    def consume(self: t.Self, version_string: str, type_headers: dict[str, str], changes: list[Change]) -> None:
         """Process sections and generate changelog file entries."""
         grouped_changes = defaultdict(list)
         for change in changes:
+            change.rendered = self._render_change(change)
             grouped_changes[change.header].append(change)
 
+        ordered_group_changes = {}
         for header in type_headers.values():
             if header not in grouped_changes:
                 continue
             # Remove processed headers to prevent rendering duplicate type -> header mappings
             changes_ = grouped_changes.pop(header)
-            self.add_section(header, changes_)
+            ordered_group_changes[header] = sorted(changes_)
 
-    @timer
-    def add_section(self: t.Self, header: str, changes: list[Change]) -> None:
-        """Add a section to changelog file."""
-        self._add_section_header(header)
-        for change in sorted(changes):
-            self._add_section_line(change)
-        self._post_section()
-
-    @timer
-    def _add_section_header(self: t.Self, header: str) -> None:
-        raise NotImplementedError
+        self._consume(version_string, ordered_group_changes)
 
     @timer
-    def _add_section_line(self: t.Self, change: Change) -> None:
+    def _consume(self: t.Self, version_string: str, group_changes: dict[str, list[Change]]) -> None:
         raise NotImplementedError
 
-    @timer
-    def _post_section(self: t.Self) -> None:
-        pass
-
     @timer
     def __str__(self: t.Self) -> str:  # noqa: D105
         content = "\n".join(self.content)
@@ -139,24 +124,21 @@ def __init__(self: t.Self, *args, **kwargs) -> None:
         )
 
     @timer
-    def _add_version(self: t.Self, version: str) -> None:
-        self.content.extend([f"## {version}", ""])
+    def _consume(self: t.Self, version_string: str, group_changes: dict[str, list[Change]]) -> None:
+        template = """## {{ version_string }}
 
-    @timer
-    def _add_section_header(self: t.Self, header: str) -> None:
-        self.content.extend([f"### {header}", ""])
+{% for header, changes in group_changes.items() -%}
+### {{ header }}
 
-    @timer
-    def _add_section_line(self: t.Self, change: Change) -> None:
-        rtemplate = Environment(loader=BaseLoader()).from_string(self._change_template.replace("\n", ""))  # noqa: S701
-
-        line = rtemplate.render(change=change)
-
-        self.content.append(line)
+{% for change in changes -%}
+{{change.rendered}}
+{% endfor %}
+{% endfor %}
+"""
+        rtemplate = Environment(loader=BaseLoader()).from_string(template)  # noqa: S701
 
-    @timer
-    def _post_section(self: t.Self) -> None:
-        self.content.append("")
+        content = rtemplate.render(group_changes=group_changes, version_string=version_string)
+        self.content = content.split("\n")[:-1]
 
 
 class RstWriter(BaseWriter):
@@ -192,25 +174,34 @@ def links(self: t.Self) -> list[str]:
         return [f".. _`{ref}`: {link}" for ref, link in sorted(self._links.items())]
 
     @timer
-    def _add_version(self: t.Self, version: str) -> None:
-        self.content.extend([version, "=" * len(version), ""])
+    def _consume(self: t.Self, version_string: str, group_changes: dict[str, list[Change]]) -> None:
+        template = """{{ version_string }}
+{{ "=" * version_string|length }}
 
-    @timer
-    def _add_section_header(self: t.Self, header: str) -> None:
-        self.content.extend([header, "-" * len(header), ""])
+{% for header, changes in group_changes.items() -%}
+{{ header }}
+{{ "-" * header|length }}
 
-    @timer
-    def _add_section_line(self: t.Self, change: Change) -> None:
-        rtemplate = Environment(loader=BaseLoader()).from_string(self._change_template.replace("\n", ""))  # noqa: S701
+{% for change in changes -%}
+{{change.rendered}}
 
-        line = rtemplate.render(change=change)
+{% endfor %}
+{% endfor %}
+"""
+        rtemplate = Environment(loader=BaseLoader()).from_string(template)  # noqa: S701
 
-        self.content.extend([line, ""])
+        content = rtemplate.render(group_changes=group_changes, version_string=version_string)
+        self.content = content.split("\n")[:-2]
+
+    @timer
+    def _render_change(self: t.Self, change: Change) -> str:
+        line = super()._render_change(change)
 
         for link in change.links:
-            line = f"{line} [`{link.text}`_]"
             self._links[link.text] = link.link
 
+        return line
+
     @timer
     def write(self: t.Self) -> str:
         """Write contents to destination."""

pyproject.toml

@@ -73,13 +73,18 @@ date_format = "- %Y-%m-%d"
 
 [[tool.changelog_gen.link_parsers]]
 target = "closes"
-pattern = "#(\\d+)$"
+pattern = '#(\d+)$'
+link = "https =//github.com/NRWLDev/changelog-gen/issues/{0}"
+
+[[tool.changelog_gen.link_parsers]]
+target = "fixes"
+pattern = '#(\d+)$'
 link = "https =//github.com/NRWLDev/changelog-gen/issues/{0}"
 
 # Fall back in case a PR does not close an issue
 [[tool.changelog_gen.link_parsers]]
 target = "Refs"
-pattern = "#(\\d+)$"
+pattern = '#(\d+)$'
 link = "https =//github.com/NRWLDev/changelog-gen/issues/{0}"
 
 [[tool.changelog_gen.link_parsers]]

tests/cli/test_config.py

@@ -64,17 +64,22 @@ def test_config_displayed(cli_runner):
 
 
 def test_post_process_config_displayed(cli_runner, config_factory):
-    config_factory(post_process={"url": "http://localhost"})
+    config_factory(post_process={"link_parser": {"target": "Refs", "link": "http://localhost"}})
     result = cli_runner.invoke(["config"])
 
     assert result.exit_code == 0
-    assert result.output.strip().endswith("""
+    assert result.output.endswith("""
 [post_process]
-url = 'http://localhost'
 verb = 'POST'
-body = '{"body": "Released on ::version::"}'
+body_template = '{"body": "Released on {{ version }}"}'
 auth_type = 'basic'
 
+[post_process.link_parser]
+target = 'Refs'
+link = 'http://localhost'
+
 [custom]
 
-[files]""")
+[files]
+
+""")