GitHub Branches

Caution

This example is for reference only. It is not extensively tested, and it not intended to be a fully-fledged Concourse resource for production pipelines. Copy and paste at your own risk.

On occasion, you may wish your resource to emit versions when something has changed, and not have a history of this to rely on. For example, imagine that we want to run a pipeline on a number of different branches of a GitHub repository. Our resource could emit a number of versions like this:

{
  "branch": "feature/new-module",
  "commit": "abcdef..."
}

But what happens when a check yields two versions from different branches? We could figure out the “latest” version by sorting by commit date, but if we are tracking multiple branches then we don’t want to ignore a commit on one just because a more recent commit has been made on another. Previous in Concourse this was “fixed” by passing version: every to the resource, but this means that if multiple commits are pushed to a branch at once, then each commit will emit a build, which may not be what we want.

What we really want is to iterate over all branches in our repository, and spin up a branch-specific pipeline for each one using the set pipeline step. This is easy to do, but we need to make sure that new pipelines are added when new branches appear, and that old pipelines are deleted when their branches are removed. We specifically require a resource which will trigger a pipeline whenever something has changed. The generic resource for this pattern is the TriggerOnChangeConcourseResource, but because the “state” is made up of several “sub versions”, it makes sense to instead utilise the MultiVersionConcourseResource, which takes care of much of the boiler plate for us, and also allow us to automatically download these subversions as JSON so that we may iterate over them.

Branch Version

For this example, each “version” will contain an encoded JSON string with all of the subversions. We only care about whether or not that version (and hence the state) has changed, and not whether these versions were linear. To start, we define the subversion schema:

@dataclass(unsafe_hash=True, order=True)
class BranchVersion(TypedVersion):
    name: str

We inherit from TypedVersion to allow us to use the dataclass() decorator and save us some lines of code. We have to specify unsafe_hash=True in order to ensure that the version instance will be hashable, otherwise we won’t be able to use it with our resource. We also need to specify order=True to allow our subversions to be sorted, as this is required to make sure the same set of subversions yield identical versions. We could have made use of the SortableVersionMixin, but given that the dataclass() decorator does this for us it isn’t necessary here.

Branch Resource

We start by inheriting from MultiVersionConcourseResource. A standard ConcourseResource will take a version class, but this time we need to pass a subversion class instead, as well as a key:

def __init__(self, owner: str, repo: str, regex: str = ".*",
             endpoint: str = "https://api.github.com") -> None:
    """
    Initialise self.

    :param owner: The owner of the repository.
    :param repo: The name of the repository.
    :param regex: An optional regex for filtering the branches. Only branches matching this
                  regex will be considered. Defaults to a regex which matches ALL branches.
    :param endpoint: The GitHub API endpoint. Defaults to the public version of GitHub.
    """
    super().__init__("branches", BranchVersion)

The resource will then construct a Version class which wraps the subversion class, and stores the list of subversions as a JSON-encoded string. The key, "branches", is used as the key in this version. The final version will look like this:

{"branches": "[{\"name\": \"issue/95\"}, {\"name\": \"master\"}, {\"name\": \"release/6.7.x\"}, {\"name\": \"version\"}, {\"name\": \"version-lts\"}]"}

We then store a compiled regex and the API route within the class. There is no need to store each parameter separately, as we can just construct the API route in one line:

def __init__(self, owner: str, repo: str, regex: str = ".*",
             endpoint: str = "https://api.github.com") -> None:
    """
    Initialise self.

    :param owner: The owner of the repository.
    :param repo: The name of the repository.
    :param regex: An optional regex for filtering the branches. Only branches matching this
                  regex will be considered. Defaults to a regex which matches ALL branches.
    :param endpoint: The GitHub API endpoint. Defaults to the public version of GitHub.
    """
    super().__init__("branches", BranchVersion)
    self.api_route = f"{endpoint}/repos/{owner}/{repo}/branches"
    self.regex = re.compile(regex)

The source of this resource would then look something like this:

source:
  owner: concourse
  repo: github-release-resource
  regex: release/.*

We only need to implement the fetch_latest_sub_versions() method; the results of this will be collected and converted into a final version by the parent class:

def fetch_latest_sub_versions(self) -> set[BranchVersion]:
    headers = {"Accept": "application/vnd.github+json"}
    response = requests.get(self.api_route, headers=headers)
    branches_info = response.json()

    try:
        branch_names = {branch_info["name"] for branch_info in branches_info}
    except TypeError as error:  # GitHub error: {"message": "..."}
        message = branches_info["message"]
        raise RuntimeError(message) from error

    return {BranchVersion(branch_name) for branch_name in branch_names
            if self.regex.fullmatch(branch_name)}

The logic required here is incredibly minimal, and we only need to return each available subversion. The returned set will then be sorted by the resource when converting it to its final version, which is why we needed the subversion to be sortable and hashable.

When the branches change (either with new ones added or existing ones removed) a new version will be emitted and the pipeline will be triggered. Users can then iterate over them in their pipeline using a combination of the across step and the set pipeline step:

- get: repo-branches
- load_var: branches
  file: branches/branches.json
- across:
  - var: branch-info
    values: ((.:branches))
  set-pipeline: branch-pipeline
  file: ...
  vars:
    branch: ((.:branch-info.name))

GitHub Branches Conclusion

The final module only requires 46 lines (including docstrings) and looks like this:

# (C) Crown Copyright GCHQ
from __future__ import annotations

from dataclasses import dataclass
import re

import requests

from concoursetools.additional import MultiVersionConcourseResource
from concoursetools.version import TypedVersion


@dataclass(unsafe_hash=True, order=True)
class BranchVersion(TypedVersion):
    name: str


class Resource(MultiVersionConcourseResource[BranchVersion]):  # type: ignore[type-var]
    def __init__(self, owner: str, repo: str, regex: str = ".*",
                 endpoint: str = "https://api.github.com") -> None:
        """
        Initialise self.

        :param owner: The owner of the repository.
        :param repo: The name of the repository.
        :param regex: An optional regex for filtering the branches. Only branches matching this
                      regex will be considered. Defaults to a regex which matches ALL branches.
        :param endpoint: The GitHub API endpoint. Defaults to the public version of GitHub.
        """
        super().__init__("branches", BranchVersion)
        self.api_route = f"{endpoint}/repos/{owner}/{repo}/branches"
        self.regex = re.compile(regex)

    def fetch_latest_sub_versions(self) -> set[BranchVersion]:
        headers = {"Accept": "application/vnd.github+json"}
        response = requests.get(self.api_route, headers=headers)
        branches_info = response.json()

        try:
            branch_names = {branch_info["name"] for branch_info in branches_info}
        except TypeError as error:  # GitHub error: {"message": "..."}
            message = branches_info["message"]
            raise RuntimeError(message) from error

        return {BranchVersion(branch_name) for branch_name in branch_names
                if self.regex.fullmatch(branch_name)}