Skip to content

Callbacks

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

Code Documentation

lso.tasks.playbook_finished_handler_factory 

playbook_finished_handler_factory(callback: str | None, job_id: str) -> Callable[[Runner], None] | None

Once Ansible runner is finished, it will call back to the specified URL in the original request.

Args: callback (str, optional): The callback URL that the Ansible runner should report to. job_id (str): The job ID of this playbook run, used for reporting.

Returns: A handler method that sends one request to the callback URL.

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

Source code in .venv/lib/python3.12/site-packages/lso/tasks.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def playbook_finished_handler_factory(callback: str | None, job_id: str) -> Callable[[Runner], None] | None:
    """Once Ansible runner is finished, it will call back to the specified URL in the original request.

    Args:
        callback (str, optional): The callback URL that the Ansible runner should report to.
        job_id (str): The job ID of this playbook run, used for reporting.

    Returns:
        A handler method that sends one request to the callback URL.

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

    """

    def _playbook_finished_handler(runner: Runner) -> None:
        playbook_output = runner.stdout.read().split("\n")
        playbook_output = [line for line in playbook_output if line.strip()]

        payload = {
            "status": runner.status,
            "job_id": job_id,
            "output": playbook_output,
            "return_code": int(str(runner.rc)),
        }

        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

    if callback:
        return _playbook_finished_handler
    return None