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.PlaybookFinishedHandler

Report a finished Ansible playbook run to the callback URL from the original request.

An instance is passed to ansible-runner as its finished_callback and invoked once when the run shuts down. It records in reported whether that report was made, so the caller can tell the run reached completion and avoid sending a second, conflicting callback if the run instead crashed before finishing.

Args: callback (str, optional): The callback URL that the Ansible runner should report to. When not set, the handler is a no-op (nothing is POSTed). job_id (str): The job ID of this playbook run, used for reporting.

Attributes: reported (bool): True once the handler has run, i.e. the playbook finished and its result callback was attempted (regardless of whether delivering it then succeeded).

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

Source code in .venv/lib/python3.14/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
113
114
115
116
117
118
119
120
121
122
123
124
125
class PlaybookFinishedHandler:
    """Report a finished Ansible playbook run to the callback URL from the original request.

    An instance is passed to `ansible-runner` as its `finished_callback` and invoked once when the run shuts down.
    It records in `reported` whether that report was made, so the caller can tell the run reached completion and
    avoid sending a second, conflicting callback if the run instead crashed before finishing.

    Args:
        callback (str, optional): The callback URL that the Ansible runner should report to. When not set, the
            handler is a no-op (nothing is POSTed).
        job_id (str): The job ID of this playbook run, used for reporting.

    Attributes:
        reported (bool): `True` once the handler has run, i.e. the playbook finished and its result callback was
            attempted (regardless of whether delivering it then succeeded).

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

    """

    def __init__(self, callback: str | None, job_id: str) -> None:
        """Store the callback URL and job ID to report on when the playbook run finishes."""
        self._callback = callback
        self._job_id = job_id
        self.reported = False

    def __call__(self, runner: Runner) -> None:
        """Send one request with the playbook result to the callback URL."""
        # Record completion before attempting delivery, so a failure while POSTing does not let the caller's
        # crash safety net fire a second callback for the same job.
        self.reported = True
        if not self._callback:
            return

        playbook_output = [line for line in runner.stdout.read().split("\n") if line.strip()]
        payload = {
            "status": runner.status,
            "job_id": self._job_id,
            "output": playbook_output,
            "return_code": int(str(runner.rc)),
        }

        response = requests.post(str(self._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

__call__ 

__call__(runner: ansible_runner.Runner) -> None

Send one request with the playbook result to the callback URL.

Source code in .venv/lib/python3.14/site-packages/lso/tasks.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def __call__(self, runner: Runner) -> None:
    """Send one request with the playbook result to the callback URL."""
    # Record completion before attempting delivery, so a failure while POSTing does not let the caller's
    # crash safety net fire a second callback for the same job.
    self.reported = True
    if not self._callback:
        return

    playbook_output = [line for line in runner.stdout.read().split("\n") if line.strip()]
    payload = {
        "status": runner.status,
        "job_id": self._job_id,
        "output": playbook_output,
        "return_code": int(str(runner.rc)),
    }

    response = requests.post(str(self._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

__init__ 

__init__(callback: str | None, job_id: str) -> None

Store the callback URL and job ID to report on when the playbook run finishes.

Source code in .venv/lib/python3.14/site-packages/lso/tasks.py 
 97
 98
 99
100
101
def __init__(self, callback: str | None, job_id: str) -> None:
    """Store the callback URL and job ID to report on when the playbook run finishes."""
    self._callback = callback
    self._job_id = job_id
    self.reported = False