Bitbucket Build Status

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.

This example will (mostly) recreate the excellent SHyx0rmZ/bitbucket-build-status-resource - specifically version 1.6.0. This is an ideal candidate to show how boiler plate code can be massively reduced to a single module concerning itself with nothing but the internal logic of your resource type.

We only care about updating the resource, and so the OutOnlyConcourseResource seems like the best route.

Build Status Version

A quick glance through the original code shows that the version contains only the build status. Since the build status must be one of three specific values, the correct thing to do is to define an Enum:

class BuildStatus(Enum):
    SUCCESSFUL = auto()
    INPROGRESS = auto()
    FAILED = auto()

The benefit of using an enum (other than it being arguably the “correct” thing to do) is that we don’t need to keep checking membership, and we can rely on the enum to raise an error upon user error. We also only need to worry about updating the enum in a single place when we need to.

Next we define the version. To make things easier, we can forget about converting the enum to a string and simply rely on TypedVersion to do it for us:

@dataclass
class Version(TypedVersion):
    build_status: BuildStatus

Build Status Resource

Next we need to think about the resource. The hardest thing about this example is that there are a number of required parameters, which are actually only required in one of two circumstances. This unfortunately means that we can’t rely on Concourse Tools to automatically catch missing parameters.

The original solves this problem through the use of driver subclasses, but since the number of conditionals remain pretty much the same we will avoid that for now, and it can always be refactored later. We can however deal with the variables in sensible groups, and rely on the right errors being thrown if the user gives the wrong parameters.

Note

The original contains a number of deprecated parameters which have not been included in this example. These could be maintained quite easily.

When we collect all resource parameters and place them in a class it can seem quite daunting:

class Resource(OutOnlyConcourseResource[Version]):

    def __init__(self, repository: str | None = None, endpoint: str | None = None,
                 username: str | None = None, password: str | None = None,
                 client_id: str | None = None, client_secret: str | None = None,
                 verify_ssl: bool = True, driver: str = "Bitbucket Server",
                 debug: bool = False) -> None:
        super().__init__(Version)

Bitbucket Authentication

The first parameter group are the authentication parameters: username, password, client_id and client_secret. We’ll ignore that the latter two don’t make sense for Bitbucket Server, and let the request fail if a user decides to try. The most Pythonic thing to do is to save an auth attribute to be used instead of the individual values, which will clutter the instance otherwise. We want a function to filter them out. If the user has passed username and password, we want to use HTTPBasicAuth:

def create_auth(username: str | None = None,
                password: str | None = None,
                client_id: str | None = None,
                client_secret: str | None = None) -> AuthBase:
    if username is not None and password is not None:
        auth: AuthBase = HTTPBasicAuth(username, password)

If the user instead passes client_id and client_secret, we instead need to use OAuth to fetch a bearer token. Neither of these are available natively in requests, and so we need to implement our own. The original resource has the following:

class BitbucketOAuth(AuthBase):
    """
    Adds the correct auth token for OAuth access to bitbucket.com.
    """
    def __init__(self, access_token: str):
        self.access_token = access_token

    def __call__(self, request: requests.Request) -> requests.Request:
        request.headers["Authorization"] = f"Bearer {self.access_token}"
        return request

We can then copy the original code for requesting the access token into an alternative constructor for ease:

class BitbucketOAuth(AuthBase):
    """
    Adds the correct auth token for OAuth access to bitbucket.com.
    """
    def __init__(self, access_token: str):
        self.access_token = access_token

    def __call__(self, request: requests.Request) -> requests.Request:
        request.headers["Authorization"] = f"Bearer {self.access_token}"
        return request

    @classmethod
    def from_client_credentials(cls, client_id: str, client_secret: str) -> BitbucketOAuth:
        token_auth = HTTPBasicAuth(client_id, client_secret)

        url = "https://bitbucket.org/site/oauth2/access_token"
        data = {"grant_type": "client_credentials"}

        token_response = requests.post(url, auth=token_auth, data=data)
        token_response.raise_for_status()
        access_token = token_response.json()["access_token"]
        return cls(access_token)

That way the create_auth function becomes quite simple:

def create_auth(username: str | None = None,
                password: str | None = None,
                client_id: str | None = None,
                client_secret: str | None = None) -> AuthBase:
    if username is not None and password is not None:
        auth: AuthBase = HTTPBasicAuth(username, password)
    elif client_id is not None and client_secret is not None:
        auth = BitbucketOAuth.from_client_credentials(client_id, client_secret)
    else:
        raise ValueError("Must set username/password or OAuth credentials")
    return auth

Bitbucket Drivers

We still need to differentiate between each type of driver, in order to know which parameters to complain about. It may be considered overkill, but I would choose to define another Enum:

class Driver(Enum):
    SERVER = "Bitbucket Server"
    CLOUD = "Bitbucket Cloud"

Again, this is slightly more resilient to change than checking a string all the time. Of course, the driver parameter to the resource has to be a string, so a lookup is required:

try:
    self.driver = Driver(driver)
except ValueError:
    possible_values = {enum.value for enum in Driver._member_map_.values()}
    raise ValueError(f"Driver must be one of the following: "

Now we can just check against the enum throughout our code. The initialiser now looks like this:

class Resource(OutOnlyConcourseResource[Version]):

    def __init__(self, repository: str | None = None, endpoint: str | None = None,
                 username: str | None = None, password: str | None = None,
                 client_id: str | None = None, client_secret: str | None = None,
                 verify_ssl: bool = True, driver: str = "Bitbucket Server",
                 debug: bool = False) -> None:
        super().__init__(Version)
        try:
            self.driver = Driver(driver)
        except ValueError:
            possible_values = {enum.value for enum in Driver._member_map_.values()}
            raise ValueError(f"Driver must be one of the following: "
                             f"{possible_values}, not {driver!r}")

        self.auth = create_auth(username, password, client_id, client_secret)

        self.repository = repository
        self.endpoint = endpoint

        self.verify_ssl = verify_ssl
        self._debug = debug

        if self.driver is Driver.SERVER:
            if endpoint is None:
                raise ValueError("Must set endpoint when using Bitbucket Server.")
            else:
                endpoint = endpoint.rstrip("/")

        if self.driver is Driver.CLOUD:
            if repository is None:
                raise ValueError("Must set repository when using Bitbucket Cloud.")

Updating the Resource

We can now write the code to update the resource, which involves implementing publish_new_version(). The method takes the usual parameters, and then the additional arguments offered by the original:

Note

The original configures a few variables via files instead of direct parameters. This is likely due to it being implemented before the load var step was a thing. This implementation replaces them with direct variables.

def publish_new_version(self, sources_dir: Path, build_metadata: BuildMetadata,
                        repository: str, build_status: str, key: str | None = None,
                        name: str | None = None, build_url: str | None = None,
                        description: str | None = None,
                        commit_hash: str | None = None) -> tuple[Version, Metadata]:

First up is debugging. The original allows the user to pass debug to the resource and have additional information be shown in the console. This required an additional function to remember, but not so with Concourse tools: we can just print!

if self._debug:
    print("--DEBUG MODE--")

This will then show up in the console when debugging has been activated. We could go one further and simplify our code with a debug function, which will also allow coloured output to be printed:

def debug(self, *args: object, colour: str = Colour.CYAN, **kwargs: Any) -> None:
    if self._debug:
        colour_print(*args, colour=colour, **kwargs)

Remember creating that Build Status enum? We should make use of it here:

try:
    status = BuildStatus[build_status]
except KeyError:
    possible_values = set(BuildStatus._member_names_)
    raise ValueError(f"Build status must be one of the following: "
                     f"{possible_values}, not {build_status!r}")

Next we need to fetch the commit hash (if it hasn’t already been set). This is more or less a direct lift from the original code, except we can make use of the fact that sources_dir is both already available to us (no shenanigans with sys.argv), but is also a Path instance, which makes the code a bit shorter:

if commit_hash is None:
    if repository is None:
        raise ValueError("Missing repository parameter.")
    repo_path = sources_dir / repository
    mercurial_path = repo_path / ".hg"
    git_path = repo_path / ".git"

    if mercurial_path.exists():
        command = ["hg", "R", str(repo_path), "log",
                   "--rev", ".", "--template", r"{node}"]
    elif git_path.exists():
        command = ["git", "-C", str(repo_path), "rev-parse", "HEAD"]
    else:
        raise RuntimeError("Cannot detect a repository.")

    commit_hash = subprocess.check_output(command).strip().decode()

Tip

If I were coding this from scratch, I would make use of the GitPython package (and the Mercurial one if it were actually stable) as part of the benefit of using Concourse Tools is having immediate access to the entire Python ecosystem, but I admit that it might seem a touch excessive given the simple command we need to run.

Next we need to figure out the default parameters for the build status. This is where Concourse Tools truly shines, as we require access to the Build Metadata. The publish_new_version() method is passed a BuildMetadata instance which contains attributes for everything that Concourse makes available to the resource, as well as some helpful convenience functions for dealing with one-off builds and instanced pipelines. In particular, it includes a build_url() method for calculating an exact URL for the build (instanced pipelines and all), which means that this original code is condensed down from 39 lines to just one:

build_url = build_url or build_metadata.build_url()

The other defaults follow in a similar fashion:

key = key or build_metadata.BUILD_JOB_NAME or f"one-off-build-{build_metadata.BUILD_ID}"

self.debug(f"Build URL: {build_url}")

description = description or f"Concourse CI build, hijack as #{build_metadata.BUILD_ID}"

if name is None:
    if build_metadata.is_one_off_build:
        name = f"One-off build #{build_metadata.BUILD_ID}"
    else:
        name = f"{build_metadata.BUILD_JOB_NAME} #{build_metadata.BUILD_NAME}"

Next, we determine the URL to which to post() our request:

if self.driver is Driver.SERVER:
    post_url = f"{self.endpoint}/rest/build-status/1.0/commits/{commit_hash}"
    if self.verify_ssl is False:
        disable_ssl_warnings()
        self.debug("SSL warnings disabled\n")

else:
    post_url = f"https://api.bitbucket.org/2.0/repositories/{self.repository}/commit/{commit_hash}/statuses/build"

Tip

Added in version 0.8.0: The format_string() method can be used to substitute build metadata into the status description in order to allow users to reference the build:

description = build_metadata.format_string(description)

Finally we submit the request and return the new version:

data = {
    "state": status.name,
    "key": key,
    "name": name,
    "url": build_url,
    "description": description,
}

self.debug(f"Set build status: {data}")

response = requests.post(post_url, json=data, auth=self.auth, verify=self.verify_ssl)

self.debug(f"Request result: {response.json()}")

version = Version(status)
metadata = {
    "HTTP Status Code": str(response.status_code),
}
return version, metadata

Note

The original doesn’t return any metadata, but this example does in order to illustrate how it works.

Conclusion

The original repository scripts cover 12 Python files and a total of 433 lines. With Concourse tools, this same code is covered in 187 lines, and a single Python file:

# (C) Crown Copyright GCHQ
from __future__ import annotations

from dataclasses import dataclass
from enum import Enum, auto
from pathlib import Path
import subprocess
from typing import Any

import requests
from requests.auth import AuthBase, HTTPBasicAuth
from urllib3 import disable_warnings as disable_ssl_warnings

from concoursetools import BuildMetadata
from concoursetools.additional import OutOnlyConcourseResource
from concoursetools.colour import Colour, colour_print
from concoursetools.typing import Metadata
from concoursetools.version import TypedVersion


class Driver(Enum):
    SERVER = "Bitbucket Server"
    CLOUD = "Bitbucket Cloud"


class BuildStatus(Enum):
    SUCCESSFUL = auto()
    INPROGRESS = auto()
    FAILED = auto()


class BitbucketOAuth(AuthBase):
    """
    Adds the correct auth token for OAuth access to bitbucket.com.
    """
    def __init__(self, access_token: str):
        self.access_token = access_token

    def __call__(self, request: requests.Request) -> requests.Request:
        request.headers["Authorization"] = f"Bearer {self.access_token}"
        return request

    @classmethod
    def from_client_credentials(cls, client_id: str, client_secret: str) -> BitbucketOAuth:
        token_auth = HTTPBasicAuth(client_id, client_secret)

        url = "https://bitbucket.org/site/oauth2/access_token"
        data = {"grant_type": "client_credentials"}

        token_response = requests.post(url, auth=token_auth, data=data)
        token_response.raise_for_status()
        access_token = token_response.json()["access_token"]
        return cls(access_token)


@dataclass
class Version(TypedVersion):
    build_status: BuildStatus


class Resource(OutOnlyConcourseResource[Version]):

    def __init__(self, repository: str | None = None, endpoint: str | None = None,
                 username: str | None = None, password: str | None = None,
                 client_id: str | None = None, client_secret: str | None = None,
                 verify_ssl: bool = True, driver: str = "Bitbucket Server",
                 debug: bool = False) -> None:
        super().__init__(Version)
        try:
            self.driver = Driver(driver)
        except ValueError:
            possible_values = {enum.value for enum in Driver._member_map_.values()}
            raise ValueError(f"Driver must be one of the following: "
                             f"{possible_values}, not {driver!r}")

        self.auth = create_auth(username, password, client_id, client_secret)

        self.repository = repository
        self.endpoint = endpoint

        self.verify_ssl = verify_ssl
        self._debug = debug

        if self.driver is Driver.SERVER:
            if endpoint is None:
                raise ValueError("Must set endpoint when using Bitbucket Server.")
            else:
                endpoint = endpoint.rstrip("/")

        if self.driver is Driver.CLOUD:
            if repository is None:
                raise ValueError("Must set repository when using Bitbucket Cloud.")

    def publish_new_version(self, sources_dir: Path, build_metadata: BuildMetadata,
                            repository: str, build_status: str, key: str | None = None,
                            name: str | None = None, build_url: str | None = None,
                            description: str | None = None,
                            commit_hash: str | None = None) -> tuple[Version, Metadata]:
        self.debug("--DEBUG MODE--")

        try:
            status = BuildStatus[build_status]
        except KeyError:
            possible_values = set(BuildStatus._member_names_)
            raise ValueError(f"Build status must be one of the following: "
                             f"{possible_values}, not {build_status!r}")

        if commit_hash is None:
            if repository is None:
                raise ValueError("Missing repository parameter.")
            repo_path = sources_dir / repository
            mercurial_path = repo_path / ".hg"
            git_path = repo_path / ".git"

            if mercurial_path.exists():
                command = ["hg", "R", str(repo_path), "log",
                           "--rev", ".", "--template", r"{node}"]
            elif git_path.exists():
                command = ["git", "-C", str(repo_path), "rev-parse", "HEAD"]
            else:
                raise RuntimeError("Cannot detect a repository.")

            commit_hash = subprocess.check_output(command).strip().decode()

        self.debug(f"Commit: {commit_hash}")

        build_url = build_url or build_metadata.build_url()

        key = key or build_metadata.BUILD_JOB_NAME or f"one-off-build-{build_metadata.BUILD_ID}"

        self.debug(f"Build URL: {build_url}")

        description = description or f"Concourse CI build, hijack as #{build_metadata.BUILD_ID}"

        if name is None:
            if build_metadata.is_one_off_build:
                name = f"One-off build #{build_metadata.BUILD_ID}"
            else:
                name = f"{build_metadata.BUILD_JOB_NAME} #{build_metadata.BUILD_NAME}"

        description = build_metadata.format_string(description)

        if self.driver is Driver.SERVER:
            post_url = f"{self.endpoint}/rest/build-status/1.0/commits/{commit_hash}"
            if self.verify_ssl is False:
                disable_ssl_warnings()
                self.debug("SSL warnings disabled\n")

        else:
            post_url = f"https://api.bitbucket.org/2.0/repositories/{self.repository}/commit/{commit_hash}/statuses/build"

        data = {
            "state": status.name,
            "key": key,
            "name": name,
            "url": build_url,
            "description": description,
        }

        self.debug(f"Set build status: {data}")

        response = requests.post(post_url, json=data, auth=self.auth, verify=self.verify_ssl)

        self.debug(f"Request result: {response.json()}")

        version = Version(status)
        metadata = {
            "HTTP Status Code": str(response.status_code),
        }
        return version, metadata

    def debug(self, *args: object, colour: str = Colour.CYAN, **kwargs: Any) -> None:
        if self._debug:
            colour_print(*args, colour=colour, **kwargs)


def create_auth(username: str | None = None,
                password: str | None = None,
                client_id: str | None = None,
                client_secret: str | None = None) -> AuthBase:
    if username is not None and password is not None:
        auth: AuthBase = HTTPBasicAuth(username, password)
    elif client_id is not None and client_secret is not None:
        auth = BitbucketOAuth.from_client_credentials(client_id, client_secret)
    else:
        raise ValueError("Must set username/password or OAuth credentials")
    return auth