Callbacks

If the callback_url was set when making the original request, LSO will return the output of the executable to this URL.

Code Documentation

lso.tasks.run_executable_proc_task

run_executable_proc_task(job_id: str, executable_path: str, args: list[str], callback: str | None) -> None

Celery task to run an arbitrary executable and notify via callback.

Executes the executable with the provided arguments and posts back the result if a callback URL is provided.

Args: job_id (str): Identifier of the job being executed. executable_path (str): Path to the executable to be executed. args (list[str]): Arguments that are passed to the executable. callback (str, optional): Callback URL for status update.

Raises: CallbackFailedError: If the callback to the external system has failed.

Source code in .venv/lib/python3.14/site-packages/lso/tasks.py

@celery.task(name=RUN_EXECUTABLE)  # type: ignore[untyped-decorator]
def run_executable_proc_task(job_id: str, executable_path: str, args: list[str], callback: str | None) -> None:
    """Celery task to run an arbitrary executable and notify via callback.

    Executes the executable with the provided arguments and posts back the result if a callback URL is provided.

    Args:
        job_id (str): Identifier of the job being executed.
        executable_path (str): Path to the executable to be executed.
        args (list[str]): Arguments that are passed to the executable.
        callback (str, optional): Callback URL for status update.

    Raises:
        CallbackFailedError: If the callback to the external system has failed.

    """
    from lso.execute import run_executable_sync  # noqa: PLC0415

    msg = f"Executing executable: {executable_path} with args: {args}, callback: {callback}"
    logger.info(msg)
    result = run_executable_sync(executable_path, args)

    if callback:
        payload = ExecutableRunResponse(
            job_id=UUID(job_id),
            result=result,
        ).model_dump(mode="json")

        response = requests.post(str(callback), json=payload, timeout=settings.REQUEST_TIMEOUT_SEC)
        try:
            response.raise_for_status()
        except HTTPError as e:
            raise CallbackFailedError(
                status_code=e.response.status_code, detail=f"{e.response.reason} for url: {e.request.url}"
            ) from e