diff --git a/.github/scripts/wave_singularity.py b/.github/scripts/wave_singularity.py index 5e97395674f..c1ffae47d0e 100755 --- a/.github/scripts/wave_singularity.py +++ b/.github/scripts/wave_singularity.py @@ -2,53 +2,86 @@ # /// script # requires-python = ">=3.10" # dependencies = [ -# "httpx", +# "requests", +# "rich", # ] # /// import logging -import httpx +import requests +import rich_click as click +from rich.logging import RichHandler +# Replace the basic logger setup with rich logging +logging.basicConfig( + level=logging.INFO, + format="%(message)s", + handlers=[ + RichHandler( + rich_tracebacks=True, + show_time=False, + markup=True, + ) + ], +) logger = logging.getLogger(__name__) +click.rich_click.SHOW_ARGUMENTS = True -image_url = "oras://community.wave.seqera.io/library/pybedtools_bedtools_htslib_pip_pypints:aa20de1f1b5ddb30" - -if image_url.startswith("oras://"): - image_url = image_url.replace("oras://", "") - -wave_api_url = "https://wave.seqera.io" -url = f"{wave_api_url}/v1alpha1/inspect" - -# if platform_pat: -# data["toweraccesstoken"] = platform_pat -# else: -# TODO -logger.warning("'platform_pat' not set, no auth to wave back end") - -try: - logger.info(f"calling image inspect at {url} for image url {image_url}") - response = httpx.post( - url=url, - json={"containerImage": image_url}, - headers={"content-type": "application/json"}, - ) - - data = response.json() - logger.debug(data) - layers = data.get("container", {}).get("manifest", {}).get("layers", []) - is_singularity = len(layers) == 1 and layers[0].get("mediaType", "").endswith(".sif") - if not is_singularity: - print(layers) - raise ValueError("not a singularity image") - if "digest" not in layers[0]: - print(layers) - raise ValueError("no 'digest' in first layer found") - - digest = layers[0]["digest"].replace("sha256:", "") - container_url = f"https://community-cr-prod.seqera.io/docker/registry/v2/blobs/sha256/{digest[:2]}/{digest}/data" - print(container_url) - -except httpx.RequestError as exc: - print(f"An error occurred while requesting {exc.request.url!r}.") - print("No singularity image for you") + +@click.command() +@click.option( + "--platform-pat", + envvar="SEQERA_ACCESS_TOKEN", + show_envvar=True, + help="Platform authentication token", +) +@click.argument("image_name") +def main(image_name, platform_pat): + """Script to return a HTTPS Singularity image URL from Seqera Containers.""" + + if image_name.startswith("oras://"): + image_name = image_name.replace("oras://", "") + + wave_api_url_base = "https://wave.seqera.io" + wave_api_url = f"{wave_api_url_base}/v1alpha1/inspect" + container_url_base = "https://community-cr-prod.seqera.io/docker/registry/v2/blobs/sha256" + + request_data = {"containerImage": image_name} + if platform_pat: + request_data["toweraccesstoken"] = platform_pat + else: + logger.debug("'--platform-pat' / '$TOWER_ACCESS_TOKEN' not set, no auth to wave back end") + + try: + logger.debug(f"Calling image inspect at {wave_api_url} for image url '{image_name}'") + response = requests.post( + url=wave_api_url, + json=request_data, + headers={"content-type": "application/json"}, + ) + + data = response.json() + logger.debug(data) + layers = data.get("container", {}).get("manifest", {}).get("layers", []) + is_singularity = len(layers) == 1 and layers[0].get("mediaType", "").endswith(".sif") + logger.debug(layers) + if not is_singularity: + raise ValueError("Not a singularity image") + if "digest" not in layers[0]: + raise ValueError("no 'digest' in first layer found") + + digest = layers[0]["digest"].replace("sha256:", "") + container_url = f"{container_url_base}/{digest[:2]}/{digest}/data" + print(container_url) + + except requests.RequestException as exc: + raise ValueError(f"An error occurred while requesting {wave_api_url}\n {exc}") + + +if __name__ == "__main__": + try: + main() + except ValueError as exc: + logger.error(f"[red]{exc}[/red]") + exit(1)