Skip to content
Snippets Groups Projects

Resolve "Add support for `artifacts` endpoint"

Merged Nitish Kumar requested to merge 13-add-support-for-artifacts-api into main
+ 29
3
@@ -89,8 +89,9 @@ def create_remote_container(
@session
def list_containers(user_input):
"""List the remote server associated with the pipeline."""
for container in _list_remote_container(user_input):
cprint(container)
if not _list_remote_container(user_input):
for container in _list_remote_container(user_input):
cprint(container)
def _list_remote_container(user_input) -> T.List[str]:
@@ -104,10 +105,35 @@ def _list_remote_container(user_input) -> T.List[str]:
server=bconfig.get_server(),
params=dict(pipeline_sha256=pipeline.checksum),
)
res.raise_for_status()
if res.status_code != 200:
return []
return res.json()["containers"].split(",")
@app.command("artifacts")
@session
def get_generated_artifacts(user_input):
# check if the containers for the corresponding pipeline exists
if not _list_remote_container(user_input):
cprint(
"Artifacts for this pipeline doesn't exist, please run the pipeline using `bitia run` first..."
)
return
pipeline = bpipeline.user_input_to_pipeline(user_input)
res = bhelper.get(
endpoint="artifacts",
server=bconfig.get_server(),
params=dict(pipeline_sha256=pipeline.checksum),
)
res.raise_for_status()
path = res._content
assert path is not None
path = path.decode()[1:-1].rstrip()
server = bconfig.get_server().rstrip("/")
cprint("Directory is being served at: ")
cprint(f"{server}/{path}")
@app.command("logs")
@session
def stream_log(user_input):
Loading