Source code for gyoza.worker.execution_pipeline.pipeline

"""Pipeline implementation."""

import traceback

import httpx

from gyoza.worker.execution_pipeline.types import ExecutionContext, Step




class Pipeline:
    """A sequence of steps executed in order.

    Parameters
    ----------
    steps : list[Step]
        Ordered list of steps to execute.
    """

    def __init__(self, steps: list[Step] | None = None) -> None:
        self._steps: list[Step] = list(steps) if steps else []



    def run(self, ctx: ExecutionContext | None = None) -> ExecutionContext:
        """Execute all steps in sequence.

        If any step raises an exception the run is marked as FAILED via the
        server API and the exception is re-raised so the caller can log it.

        The one exception is a lost connection to the server
        (``httpx.TransportError``), which is not reported: it says nothing about
        the op or its data, and a FAILED written for it ended the run on its
        first attempt. The attempt is left as it is instead; once the worker
        stops listing it, the reaper hands the run to a fresh attempt, as it does
        for any run its worker dropped. ``HTTPStatusError`` is a real answer from
        the server and still fails the run.

        Parameters
        ----------
        ctx : ExecutionContext | None
            Shared context. Created if not provided.

        Returns
        -------
        ExecutionContext
            The context after all steps have been applied.

        Raises
        ------
        Exception
            Whatever exception the failing step raised.
        """
        if ctx is None:
            ctx = ExecutionContext()

        try:
            for step in self._steps:
                ctx = step(ctx)
        except httpx.TransportError:
            raise
        except Exception as exc:
            from gyoza.client import gyoza_client  # noqa: PLC0415

            _MAX_MSG_LEN = 10000
            msg = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()}"
            try:
                gyoza_client.add_event(
                    run_id=ctx.oprun_id,
                    attempt=ctx.attempt,
                    worker_id=ctx.worker_id,
                    event_type="FAILED",
                    msg=msg[:_MAX_MSG_LEN],
                    payload={"error_message": f"{type(exc).__name__}: {exc}"[:500]},
                )
            except Exception:
                pass
            raise

        return ctx