Skip to content

Instantly share code, notes, and snippets.

@makmanalp
Last active January 1, 2019 15:50
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save makmanalp/1e5f5c27bef262f20e603e543712b414 to your computer and use it in GitHub Desktop.
Save makmanalp/1e5f5c27bef262f20e603e543712b414 to your computer and use it in GitHub Desktop.
Quick and dirty ansible module for fetching CircleCI build artifacts (latest on a branch, or by build num & git SHA)
#!/usr/bin/env python
import requests
from ansible.module_utils.basic import AnsibleModule
import traceback
try:
from urllib.parse import quote
except ImportError:
from urllib import quote
def _get_latest_build_no(circleci_url_prefix, circleci_token, user, project,
branch, latest_build_filter="successful",
version_control="github", workflows_job_name=None):
latest_successful_build_url = "{circleci_url_prefix}/project/{version_control}/{user}/{project}/tree/{branch}?filter={latest_build_filter}&circle-token={circleci_token}".format(
circleci_url_prefix=circleci_url_prefix,
version_control=version_control, user=user, project=project,
branch=quote(branch), latest_build_filter=latest_build_filter,
circleci_token=circleci_token,
)
response = requests.get(latest_successful_build_url)
assert response.status_code == 200
if not workflows_job_name:
return response.json()[0]["build_num"]
else:
return list(
filter(
lambda x: x["workflows"]["job_name"] == workflows_job_name if "workflows" in x else False,
response.json()
))[0]["build_num"]
def _get_build_details(circleci_url_prefix, circleci_token, user, project,
build_num, version_control="github"):
build_url = "{circleci_url_prefix}/project/{version_control}/{user}/{project}/{build_num}?circle-token={circleci_token}".format(
circleci_url_prefix=circleci_url_prefix,
version_control=version_control, user=user, project=project,
build_num=build_num, circleci_token=circleci_token,
)
response = requests.get(build_url)
assert response.status_code == 200
return response.json()
def get_build_artifact(circleci_url_prefix, circleci_token, user, project,
branch, build_num=None, commit_hash=None,
artifact_path=None, latest_build_filter="successful",
version_control="github", workflows_job_name=None):
if build_num is None:
build_num = _get_latest_build_no(circleci_url_prefix, circleci_token,
user, project, branch,
latest_build_filter, version_control,
workflows_job_name)
else:
details = _get_build_details(circleci_url_prefix, circleci_token, user,
project, build_num, version_control)
assert details["vcs_revision"] == commit_hash
artifact_url = "{circleci_url_prefix}/project/{version_control}/{user}/{project}/{build_num}/artifacts?circle-token={circleci_token}".format(
circleci_url_prefix=circleci_url_prefix,
version_control=version_control, user=user, project=project,
build_num=build_num, circleci_token=circleci_token,
)
response = requests.get(artifact_url).json()
if artifact_path:
artifact_list = list(
filter(
lambda x: x["path"] == artifact_path,
response
))
if len(artifact_list) == 0:
return None
else:
return artifact_list[0]["url"]
else:
return [artifact["url"] for artifact in response]
def main():
arg_spec = {
"circleci_token": {"required": True, "type": "str"},
"circleci_url_prefix": {"required": False, "type": "str", "default": "https://circleci.com/api/v1.1"},
"user": {"required": True, "type": "str"},
"project": {"required": True, "type": "str"},
"branch": {"required": False, "type": "str"},
"build_num": {"required": False, "type": "int", "default": None},
"commit_hash": {"required": False, "type": "str", "default": None},
"artifact_path": {"required": False, "type": "str", "default": None},
"latest_build_filter": {"required": False, "type": "str", "default": "successful"},
"version_control": {"required": False, "type": "str", "default": "github"},
"workflows_job_name": {"required": False, "type": "str", "default": None},
}
m = AnsibleModule(argument_spec=arg_spec)
try:
artifact = get_build_artifact(**m.params)
except Exception as ex:
tb_str = ''.join(traceback.format_exception(etype=type(ex), value=ex,
tb=ex.__traceback__))
m.exit_json(failed=True, error_message=tb_str, artifact_url=None)
m.exit_json(changed=True, artifact_url=artifact)
if __name__ == "__main__":
main()
@makmanalp
Copy link
Author

More on fetching builds by commit hash / revision: https://twitter.com/FelicianoTech/status/1027242152965877760

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment