File indexing completed on 2025-01-18 09:15:42
0001 import io
0002 from pathlib import Path
0003 from zipfile import ZipFile
0004
0005 import requests
0006
0007
0008 def download_artifact(workflow, artifact_name, token=None, click=None):
0009 artifacts = [artifact for artifact in workflow.get_artifacts() if artifact.name == artifact_name]
0010 if not artifacts:
0011 if click is not None:
0012 click.secho(f"Can not obtain {artifact_name}", fg="red", err=True)
0013 if workflow.get_artifacts().totalCount:
0014 click.secho(f"Available artifacts:", fg="red", err=True)
0015 for artifact in workflow.get_artifacts():
0016 click.echo(artifact.name)
0017 else:
0018 click.secho(f"No artifacts available", fg="red", err=True)
0019 return None
0020
0021 outdir = Path(workflow.created_at.isoformat().replace(":", "-") + "_" + workflow.head_sha)
0022 outpath = outdir / artifact_name
0023 if outpath.exists():
0024 return outpath
0025
0026 if not outpath.parent.exists():
0027 outdir.mkdir()
0028
0029 artifact, = artifacts
0030 req = requests.get(artifact.archive_download_url, headers={"Authorization": f"token {token}"} if token else {})
0031 zfp = ZipFile(io.BytesIO(req.content))
0032 if artifact_name in zfp.namelist():
0033 zip_filename = artifact_name
0034 elif len(zfp.namelist()) == 1:
0035 zip_filename, = zfp.namelist()
0036 if click is not None:
0037 click.secho(f"Can't locate {artifact_name} in the artifact ZIP archive, using {zip_filename} instead", fg="orange", err=True)
0038 else:
0039 if click is not None:
0040 click.secho(f"Can't locate {artifact_name} in the artifact ZIP archive", fg="red", err=True)
0041 return None
0042 with zfp.open(zip_filename) as fp_zip:
0043 with open(outpath, "wb") as fp_out:
0044 fp_out.write(fp_zip.read())
0045
0046 return outpath
