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):
def __init__(self, repository: Optional[str] = None, endpoint: Optional[str] = None,
username: Optional[str] = None, password: Optional[str] = None,
client_id: Optional[str] = None, client_secret: Optional[str] = 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: Optional[str] = None,
password: Optional[str] = None,
client_id: Optional[str] = None,
client_secret: Optional[str] = None) -> AuthBase:
if username is not None and password is not None:
auth = 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):
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: Optional[str] = None,
password: Optional[str] = None,
client_id: Optional[str] = None,
client_secret: Optional[str] = None) -> AuthBase:
if username is not None and password is not None:
auth = 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):
def __init__(self, repository: Optional[str] = None, endpoint: Optional[str] = None,
username: Optional[str] = None, password: Optional[str] = None,
client_id: Optional[str] = None, client_secret: Optional[str] = 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: pathlib.Path, build_metadata: BuildMetadata,
repository: str, build_status: str, key: Optional[str] = None,
name: Optional[str] = None, build_url: Optional[str] = None,
description: Optional[str] = None,
commit_hash: Optional[str] = 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, colour=Colour.CYAN, **kwargs):
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"
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 172 lines, and a single Python file:
1# (C) Crown Copyright GCHQ
2from dataclasses import dataclass
3from enum import Enum, auto
4import pathlib
5import subprocess
6from typing import Optional, Tuple
7
8import requests
9from requests.auth import AuthBase, HTTPBasicAuth
10from urllib3 import disable_warnings as disable_ssl_warnings
11
12from concoursetools import BuildMetadata
13from concoursetools.additional import OutOnlyConcourseResource
14from concoursetools.colour import Colour, colour_print
15from concoursetools.typing import Metadata
16from concoursetools.version import TypedVersion
17
18
19class Driver(Enum):
20 SERVER = "Bitbucket Server"
21 CLOUD = "Bitbucket Cloud"
22
23
24class BuildStatus(Enum):
25 SUCCESSFUL = auto()
26 INPROGRESS = auto()
27 FAILED = auto()
28
29
30class BitbucketOAuth(AuthBase):
31 """
32 Adds the correct auth token for OAuth access to bitbucket.com.
33 """
34 def __init__(self, access_token: str):
35 self.access_token = access_token
36
37 def __call__(self, request: requests.Request) -> requests.Request:
38 request.headers["Authorization"] = f"Bearer {self.access_token}"
39 return request
40
41 @classmethod
42 def from_client_credentials(cls, client_id: str, client_secret: str):
43 token_auth = HTTPBasicAuth(client_id, client_secret)
44
45 url = "https://bitbucket.org/site/oauth2/access_token"
46 data = {"grant_type": "client_credentials"}
47
48 token_response = requests.post(url, auth=token_auth, data=data)
49 token_response.raise_for_status()
50 access_token = token_response.json()["access_token"]
51 return cls(access_token)
52
53
54@dataclass
55class Version(TypedVersion):
56 build_status: BuildStatus
57
58
59class Resource(OutOnlyConcourseResource):
60
61 def __init__(self, repository: Optional[str] = None, endpoint: Optional[str] = None,
62 username: Optional[str] = None, password: Optional[str] = None,
63 client_id: Optional[str] = None, client_secret: Optional[str] = None,
64 verify_ssl: bool = True, driver: str = "Bitbucket Server",
65 debug: bool = False) -> None:
66 super().__init__(Version)
67 try:
68 self.driver = Driver(driver)
69 except ValueError:
70 possible_values = {enum.value for enum in Driver._member_map_.values()}
71 raise ValueError(f"Driver must be one of the following: "
72 f"{possible_values}, not {driver!r}")
73
74 self.auth = create_auth(username, password, client_id, client_secret)
75
76 self.repository = repository
77 self.endpoint = endpoint
78
79 self.verify_ssl = verify_ssl
80 self._debug = debug
81
82 if self.driver is Driver.SERVER:
83 if endpoint is None:
84 raise ValueError("Must set endpoint when using Bitbucket Server.")
85 else:
86 endpoint = endpoint.rstrip("/")
87
88 if self.driver is Driver.CLOUD:
89 if repository is None:
90 raise ValueError("Must set repository when using Bitbucket Cloud.")
91
92 def publish_new_version(self, sources_dir: pathlib.Path, build_metadata: BuildMetadata,
93 repository: str, build_status: str, key: Optional[str] = None,
94 name: Optional[str] = None, build_url: Optional[str] = None,
95 description: Optional[str] = None,
96 commit_hash: Optional[str] = None) -> Tuple[Version, Metadata]:
97 self.debug("--DEBUG MODE--")
98
99 try:
100 status = BuildStatus[build_status]
101 except KeyError:
102 possible_values = set(BuildStatus._member_names_)
103 raise ValueError(f"Build status must be one of the following: "
104 f"{possible_values}, not {build_status!r}")
105
106 if commit_hash is None:
107 if repository is None:
108 raise ValueError("Missing repository parameter.")
109 repo_path = sources_dir / repository
110 mercurial_path = repo_path / ".hg"
111 git_path = repo_path / ".git"
112
113 if mercurial_path.exists():
114 command = ["hg", "R", str(repo_path), "log",
115 "--rev", ".", "--template", r"{node}"]
116 elif git_path.exists():
117 command = ["git", "-C", str(repo_path), "rev-parse", "HEAD"]
118 else:
119 raise RuntimeError("Cannot detect a repository.")
120
121 commit_hash = subprocess.check_output(command).strip().decode()
122
123 self.debug(f"Commit: {commit_hash}")
124
125 build_url = build_url or build_metadata.build_url()
126
127 key = key or build_metadata.BUILD_JOB_NAME or f"one-off-build-{build_metadata.BUILD_ID}"
128
129 self.debug(f"Build URL: {build_url}")
130
131 description = description or f"Concourse CI build, hijack as #{build_metadata.BUILD_ID}"
132
133 if name is None:
134 if build_metadata.is_one_off_build:
135 name = f"One-off build #{build_metadata.BUILD_ID}"
136 else:
137 name = f"{build_metadata.BUILD_JOB_NAME} #{build_metadata.BUILD_NAME}"
138
139 if self.driver is Driver.SERVER:
140 post_url = f"{self.endpoint}/rest/build-status/1.0/commits/{commit_hash}"
141 if self.verify_ssl is False:
142 disable_ssl_warnings()
143 self.debug("SSL warnings disabled\n")
144
145 else:
146 post_url = f"https://api.bitbucket.org/2.0/repositories/{self.repository}/commit/{commit_hash}/statuses/build"
147
148 data = {
149 "state": status.name,
150 "key": key,
151 "name": name,
152 "url": build_url,
153 "description": description,
154 }
155
156 self.debug(f"Set build status: {data}")
157
158 response = requests.post(post_url, json=data, auth=self.auth, verify=self.verify_ssl)
159
160 self.debug(f"Request result: {response.json()}")
161
162 version = Version(status)
163 metadata = {
164 "HTTP Status Code": str(response.status_code),
165 }
166 return version, metadata
167
168 def debug(self, *args, colour=Colour.CYAN, **kwargs):
169 if self._debug:
170 colour_print(*args, colour=colour, **kwargs)
171
172
173def create_auth(username: Optional[str] = None,
174 password: Optional[str] = None,
175 client_id: Optional[str] = None,
176 client_secret: Optional[str] = None) -> AuthBase:
177 if username is not None and password is not None:
178 auth = HTTPBasicAuth(username, password)
179 elif client_id is not None and client_secret is not None:
180 auth = BitbucketOAuth.from_client_credentials(client_id, client_secret)
181 else:
182 raise ValueError("Must set username/password or OAuth credentials")
183 return auth