Skip to content
Snippets Groups Projects
Commit 3ee9da42 authored by dilawar's avatar dilawar :ant:
Browse files

Merge branch '13-add-support-for-artifacts-api' into 'main'

Resolve "Add support for `artifacts` endpoint"

Closes #13

See merge request !19
parents ffd1b2f1 d35f1120
No related branches found
No related tags found
1 merge request!19Resolve "Add support for `artifacts` endpoint"
Pipeline #6158 passed with stages
in 2 minutes and 40 seconds
...@@ -89,8 +89,9 @@ def create_remote_container( ...@@ -89,8 +89,9 @@ def create_remote_container(
@session @session
def list_containers(user_input): def list_containers(user_input):
"""List the remote server associated with the pipeline.""" """List the remote server associated with the pipeline."""
for container in _list_remote_container(user_input): if not _list_remote_container(user_input):
cprint(container) for container in _list_remote_container(user_input):
cprint(container)
def _list_remote_container(user_input) -> T.List[str]: def _list_remote_container(user_input) -> T.List[str]:
...@@ -104,10 +105,35 @@ 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(), server=bconfig.get_server(),
params=dict(pipeline_sha256=pipeline.checksum), params=dict(pipeline_sha256=pipeline.checksum),
) )
res.raise_for_status() if res.status_code != 200:
return []
return res.json()["containers"].split(",") 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") @app.command("logs")
@session @session
def stream_log(user_input): def stream_log(user_input):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment