build_artifacts/v2/v2.5/v2.5.0/dirs/etc/sagemaker-ui/workflows/workflow_client.py [6:49]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
JUPYTERLAB_URL = "http://default:8888/jupyterlab/default/"
WORKFLOWS_API_ENDPOINT = "api/sagemaker/workflows"
TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S.%f%z"

def _validate_response(function_name: str, response: requests.Response):
    if response.status_code == 200:
        return response
    else:
        raise RuntimeError(
            f"{function_name} returned {response.status_code}: {str(response.content)}"
        )


def update_local_runner_status(
    session: requests.Session, status: str, detailed_status: Optional[str] = None, **kwargs
):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/update-local-runner-status",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={
            "timestamp": datetime.now(timezone.utc).strftime(TIMESTAMP_FORMAT),
            "status": status,
            "detailed_status": detailed_status,
        },
    )
    return _validate_response("UpdateLocalRunner", response)


def start_local_runner(session: requests.Session, **kwargs):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/start-local-runner",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={},
    )
    return _validate_response("StartLocalRunner", response)


def stop_local_runner(session: requests.Session, **kwargs):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/stop-local-runner",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={},
    )
    return _validate_response("StopLocalRunner", response)
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



build_artifacts/v2/v2.6/v2.6.0/dirs/etc/sagemaker-ui/workflows/workflow_client.py [7:50]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
JUPYTERLAB_URL = "http://default:8888/jupyterlab/default/"
WORKFLOWS_API_ENDPOINT = "api/sagemaker/workflows"
TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S.%f%z"

def _validate_response(function_name: str, response: requests.Response):
    if response.status_code == 200:
        return response
    else:
        raise RuntimeError(
            f"{function_name} returned {response.status_code}: {str(response.content)}"
        )


def update_local_runner_status(
    session: requests.Session, status: str, detailed_status: Optional[str] = None, **kwargs
):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/update-local-runner-status",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={
            "timestamp": datetime.now(timezone.utc).strftime(TIMESTAMP_FORMAT),
            "status": status,
            "detailed_status": detailed_status,
        },
    )
    return _validate_response("UpdateLocalRunner", response)


def start_local_runner(session: requests.Session, **kwargs):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/start-local-runner",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={},
    )
    return _validate_response("StartLocalRunner", response)


def stop_local_runner(session: requests.Session, **kwargs):
    response = session.post(
        url=JUPYTERLAB_URL + WORKFLOWS_API_ENDPOINT + "/stop-local-runner",
        headers={"X-Xsrftoken": session.cookies.get_dict()["_xsrf"]},
        json={},
    )
    return _validate_response("StopLocalRunner", response)
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -