sematic
sematic copied to clipboard
Add public API for blocking on a run until it's complete
People may want to launch a run and then wait to see if it succeeds (ex: if using a Sematic pipeline in CI). This can be done as follows:
import time
from sematic import api_client
from sematic.abstract_future import FutureState
def block_until_done(run_id: str, poll_interval_seconds: int):
keep_going = True
while keep_going:
run = api_client.get_run(run_id)
state = FutureState[run.future_state]
keep_going = not state.is_terminal()
time.sleep(poll_interval_seconds)
if state != FutureState.RESOLVED:
raise RuntimeError(f"Run {run_id} finished in state {state}")
However, this uses non-public APIs. We should expose something like block_until_done
as a public API.