Context Managers

Bind a :class:PipelineReport to the current context.

Lets nested helpers (e.g., :func:pipeline_step, :func:pipeline_file) discover the active pipeline via a thread/async-safe context variable, so you don’t have to pass pr explicitly.

Parameters:

Name	Type	Description	Default
`pr`	`PipelineReport`	The pipeline to bind for the duration of the context.	required

Yields:

Type	Description
`PipelineReport`	The same report object (convenience).

Examples:

>>> with bind_pipeline(report):
...     # inside, helpers can omit `pr`
...     ...

Source code in src/pipeline_watcher/core.py

@contextmanager
def bind_pipeline(pr: "PipelineReport"):
    """Bind a :class:`PipelineReport` to the current context.

    Lets nested helpers (e.g., :func:`pipeline_step`, :func:`pipeline_file`)
    discover the active pipeline via a thread/async-safe context variable,
    so you don’t have to pass ``pr`` explicitly.

    Parameters
    ----------
    pr : PipelineReport
        The pipeline to bind for the duration of the context.

    Yields
    ------
    PipelineReport
        The same report object (convenience).

    Examples
    --------
    >>> with bind_pipeline(report):
    ...     # inside, helpers can omit `pr`
    ...     ...
    """
    token = _current_pipeline_report.set(pr)
    try:
        yield pr
    finally:
        _current_pipeline_report.reset(token)

Context manager for a batch-level step.

Creates a :class:StepReport, times it, captures exceptions (optional re-raise), and appends it to the provided or bound pipeline. Also supports updating the pipeline's progress banner on enter/exit.

Parameters:

Name	Type	Description	Default
`pr`	`PipelineReport or None`	Pipeline to append the step to. If `None`, uses the pipeline bound via :func:`bind_pipeline`. If neither is available, the step is still yielded but not appended.	required
`id`	`str`	Machine-friendly step id (e.g., `"validate_manifest"`).	required
`label`	`str`	Human-friendly label for UI display.	`None`
`banner_stage`	`str`	Stage label to set on the pipeline banner after appending the step (or on enter if `set_stage_on_enter=True`). Defaults to `id` when provided as `None`.	`None`
`banner_percent`	`int`	Percent to set on the banner after appending. If `None`, leaves the current percent unchanged.	`None`
`banner_message`	`str`	Message to set on the banner (falls back to existing message if `None`).	`None`
`set_stage_on_enter`	`bool`	If `True`, set the banner's `stage`/`message` when entering the context (percent remains unchanged on enter).	`False`
`raise_on_exception`	`bool`	If `True`, re-raise the exception after recording it on the step. If `False`, swallow after recording so the pipeline can continue.	`False`
`save_on_exception`	`bool`	If an exception occurs and a pipeline is available, attempt to save the pipeline JSON immediately (best-effort).	`True`
`output_path_override`	`str or Path or None`	When saving on exception, write to this path instead of `pr.output_path` if provided.	`None`

Yields:

Type	Description
`StepReport`	The live step report to populate within the `with` block.

Notes

Exceptions inside the block are recorded as:
errors += ["{Type}: {message}"]
metadata['traceback'] = traceback.format_exc()
status set to FAILED via st.fail(...)
metadata['duration_ms'] is recorded on exit.
If a pipeline is available, the step is finalized via st.end(), appended with :meth:PipelineReport.append_step, and the banner is optionally updated.

Examples:

Minimal with bound pipeline:

>>> with bind_pipeline(report):
...     with pipeline_step(None, "index", label="Index batch") as st:
...         st.add_check("manifest_present", ok=True)

Update the banner and save immediately on error:

>>> with pipeline_step(
...     report, "extract",
...     banner_stage="extract",
...     banner_percent=40,
...     banner_message="Extracting data…",
...     save_on_exception=True,
... ):
...     risky_operation()

Source code in src/pipeline_watcher/core.py

@contextmanager
def pipeline_step(
    pr: Optional["PipelineReport"],
    id: str,
    *,
    label: str | None = None,
    banner_stage: str | None = None,
    banner_percent: int | None = None,
    banner_message: str | None = None,
    set_stage_on_enter: bool = False,
    raise_on_exception: bool = False,
    save_on_exception: bool = True,
    output_path_override: str | Path | None = None,
):
    """Context manager for a **batch-level** step.

    Creates a :class:`StepReport`, times it, captures exceptions (optional
    re-raise), and appends it to the provided or bound pipeline. Also supports
    updating the pipeline's progress banner on enter/exit.

    Parameters
    ----------
    pr : PipelineReport or None
        Pipeline to append the step to. If ``None``, uses the pipeline bound
        via :func:`bind_pipeline`. If neither is available, the step is still
        yielded but not appended.
    id : str
        Machine-friendly step id (e.g., ``"validate_manifest"``).
    label : str, optional
        Human-friendly label for UI display.
    banner_stage : str, optional
        Stage label to set on the pipeline banner after appending the step
        (or on enter if ``set_stage_on_enter=True``). Defaults to ``id`` when
        provided as ``None``.
    banner_percent : int, optional
        Percent to set on the banner after appending. If ``None``, leaves the
        current percent unchanged.
    banner_message : str, optional
        Message to set on the banner (falls back to existing message if ``None``).
    set_stage_on_enter : bool, default False
        If ``True``, set the banner's ``stage``/``message`` when entering the
        context (percent remains unchanged on enter).
    raise_on_exception : bool, default False
        If ``True``, re-raise the exception after recording it on the step.
        If ``False``, swallow after recording so the pipeline can continue.
    save_on_exception : bool, default True
        If an exception occurs and a pipeline is available, attempt to save
        the pipeline JSON immediately (best-effort).
    output_path_override : str or Path or None, optional
        When saving on exception, write to this path instead of
        ``pr.output_path`` if provided.

    Yields
    ------
    StepReport
        The live step report to populate within the ``with`` block.

    Notes
    -----
    - Exceptions inside the block are recorded as:
      - ``errors += [\"{Type}: {message}\"]``
      - ``metadata['traceback'] = traceback.format_exc()``
      - status set to ``FAILED`` via ``st.fail(...)``
    - ``metadata['duration_ms']`` is recorded on exit.
    - If a pipeline is available, the step is finalized via ``st.end()``,
      appended with :meth:`PipelineReport.append_step`, and the banner is
      optionally updated.

    Examples
    --------
    Minimal with bound pipeline:

    >>> with bind_pipeline(report):
    ...     with pipeline_step(None, "index", label="Index batch") as st:
    ...         st.add_check("manifest_present", ok=True)

    Update the banner and save immediately on error:

    >>> with pipeline_step(
    ...     report, "extract",
    ...     banner_stage="extract",
    ...     banner_percent=40,
    ...     banner_message="Extracting data…",
    ...     save_on_exception=True,
    ... ):
    ...     risky_operation()
    """
    pr = pr or _current_pipeline_report.get()
    st = StepReport.begin(id, label=label)
    t0 = time.perf_counter()

    if pr is not None and set_stage_on_enter:
        pr.set_progress(stage=banner_stage or id, percent=pr.percent, message=banner_message or pr.message)

    exc: BaseException | None = None
    try:
        yield st
        st.end()
    except BaseException as e:
        exc = e
        st.errors.append(f"{type(e).__name__}: {e}")
        st.metadata["traceback"] = traceback.format_exc()
        st.fail("Unhandled exception")
    finally:
        st.metadata["duration_ms"] = round((time.perf_counter() - t0) * 1000, 3)

        if pr is not None:
            try:
                pr.append_step(st)
                if any(v is not None for v in (banner_stage, banner_percent, banner_message)):
                    pr.set_progress(
                        stage=banner_stage or (pr.stage or id),
                        percent=pr.percent if banner_percent is None else banner_percent,
                        message=banner_message or pr.message,
                    )
            finally:
                if exc and save_on_exception:
                    try:
                        print(f"trying to save to {output_path_override or pr.output_path}")
                        pr.save(output_path_override or pr.output_path)
                    except Exception as save_err:
                        st.warnings.append(f"save failed: {save_err!r}")

        if exc and raise_on_exception:
            raise exc

Context manager for a per-file processing block.

Creates a :class:FileReport, times it, captures exceptions (optional re-raise), appends it to the provided or bound pipeline, and optionally updates the pipeline banner on enter/exit.

Parameters:

Name	Type	Description	Default
`pr`	`PipelineReport or None`	Pipeline to append the file report to. If `None`, uses the pipeline bound via :func:`bind_pipeline`. If neither available, the file report is yielded but not appended.	required
`file_id`	`str`	Stable identifier for the file (preferably unique in the batch).	required
`path`	`str or None`	Display/source info for UIs.	`None`
`name`	`str or None`	Display/source info for UIs.	`None`
`size_bytes`	`int or None`	File size hint.	`None`
`mime_type`	`str or None`	MIME type hint for UIs.	`None`
`metadata`	`dict or None`	Extra metadata merged into the file report.	`None`
`set_stage_on_enter`	`bool`	If `True`, set the pipeline banner’s `stage` when entering the context (percent/message left as-is on enter).	`False`
`banner_stage`	`str or None`	Stage to set when updating the banner (on enter if `set_stage_on_enter=True` and/or on exit when appending). Defaults to `name` or `file_id` on enter; leaves stage unchanged on exit if `None`.	`None`
`banner_percent_on_exit`	`int or None`	Percent to set on the banner after appending the file. If `None`, keeps current percent.	`None`
`banner_message_on_exit`	`str or None`	Message to set on the banner after appending. If `None`, keeps current message.	`None`
`raise_on_exception`	`bool`	If `True`, re-raise after recording error; if `False`, swallow so the pipeline can continue.	`False`
`save_on_exception`	`bool`	If an exception occurs and a pipeline is available, attempt to save the pipeline JSON immediately (best effort).	`True`
`output_path_override`	`str or Path or None`	When saving on exception, write here instead of `pr.output_path`.	`None`

Yields:

Type	Description
`FileReport`	The live file report to populate inside the `with` block.

Notes

Exceptions are recorded on the file report:
errors += ["{Type}: {message}"]
metadata['traceback'] = traceback.format_exc()
status set via fail("Unhandled exception while processing file")
metadata['duration_ms'] is recorded on exit.
On normal completion, the file is marked SUCCESS and finalized.

Examples:

>>> with bind_pipeline(report):
...     with pipeline_file(None, file_id="f1", path="inputs/a.pdf", name="a.pdf") as fr:
...         fr.add_completed_step("Verified file exists")

Source code in src/pipeline_watcher/core.py

@contextmanager
def pipeline_file(
    pr: Optional["PipelineReport"],
    *,
    file_id: str,
    path: str | None = None,
    name: str | None = None,
    size_bytes: int | None = None,
    mime_type: str | None = None,
    metadata: dict | None = None,
    set_stage_on_enter: bool = False,
    banner_stage: str | None = None,
    banner_percent_on_exit: int | None = None,
    banner_message_on_exit: str | None = None,
    raise_on_exception: bool = False,
    save_on_exception: bool = True,
    output_path_override: str | Path | None = None
):
    """Context manager for a **per-file** processing block.

    Creates a :class:`FileReport`, times it, captures exceptions (optional
    re-raise), appends it to the provided or bound pipeline, and optionally
    updates the pipeline banner on enter/exit.

    Parameters
    ----------
    pr : PipelineReport or None
        Pipeline to append the file report to. If ``None``, uses the pipeline
        bound via :func:`bind_pipeline`. If neither available, the file report
        is yielded but not appended.
    file_id : str
        Stable identifier for the file (preferably unique in the batch).
    path, name : str or None, optional
        Display/source info for UIs.
    size_bytes : int or None, optional
        File size hint.
    mime_type : str or None, optional
        MIME type hint for UIs.
    metadata : dict or None, optional
        Extra metadata merged into the file report.
    set_stage_on_enter : bool, default False
        If ``True``, set the pipeline banner’s ``stage`` when entering the
        context (percent/message left as-is on enter).
    banner_stage : str or None, optional
        Stage to set when updating the banner (on enter if
        ``set_stage_on_enter=True`` and/or on exit when appending). Defaults to
        ``name`` or ``file_id`` on enter; leaves stage unchanged on exit if
        ``None``.
    banner_percent_on_exit : int or None, optional
        Percent to set on the banner **after** appending the file. If ``None``,
        keeps current percent.
    banner_message_on_exit : str or None, optional
        Message to set on the banner **after** appending. If ``None``, keeps
        current message.
    raise_on_exception : bool, default False
        If ``True``, re-raise after recording error; if ``False``, swallow so
        the pipeline can continue.
    save_on_exception : bool, default True
        If an exception occurs and a pipeline is available, attempt to save the
        pipeline JSON immediately (best effort).
    output_path_override : str or Path or None, optional
        When saving on exception, write here instead of ``pr.output_path``.

    Yields
    ------
    FileReport
        The live file report to populate inside the ``with`` block.

    Notes
    -----
    - Exceptions are recorded on the file report:
      - ``errors += [\"{Type}: {message}\"]``
      - ``metadata['traceback'] = traceback.format_exc()``
      - status set via ``fail("Unhandled exception while processing file")``
    - ``metadata['duration_ms']`` is recorded on exit.
    - On normal completion, the file is marked ``SUCCESS`` and finalized.

    Examples
    --------
    >>> with bind_pipeline(report):
    ...     with pipeline_file(None, file_id="f1", path="inputs/a.pdf", name="a.pdf") as fr:
    ...         fr.add_completed_step("Verified file exists")
    """
    pr = pr or _current_pipeline_report.get()

    fr = FileReport.begin(file_id=file_id, path=path, name=name)
    if size_bytes is not None: fr.size_bytes = size_bytes
    if mime_type  is not None: fr.mime_type  = mime_type
    if metadata:               fr.metadata.update(metadata)

    t0 = time.perf_counter()

    if pr is not None and set_stage_on_enter:
        pr.set_progress(stage=banner_stage or (name or file_id),
                        percent=pr.percent,
                        message=pr.message)

    exc: BaseException | None = None
    try:
        yield fr
        fr.succeed()  # only runs if the block completes successfully
    except BaseException as e:
        exc = e
        fr.errors.append(f"{type(e).__name__}: {e}")
        fr.metadata["traceback"] = traceback.format_exc()
        fr.fail("Unhandled exception while processing file")
    finally:
        fr.metadata["duration_ms"] = round((time.perf_counter() - t0) * 1000, 3)
        fr.end()
        if pr is not None:
            try:
                pr.append_file(fr)
                if banner_stage or banner_percent_on_exit is not None or banner_message_on_exit:
                    pr.set_progress(
                        stage=banner_stage or pr.stage,
                        percent=pr.percent if banner_percent_on_exit is None else banner_percent_on_exit,
                        message=banner_message_on_exit or pr.message,
                    )
            finally:
                if exc and save_on_exception:
                    try:
                        pr.save(output_path_override or pr.output_path)
                    except Exception as save_err:
                        fr.warnings.append(f"save failed: {save_err!r}")

        if exc and raise_on_exception:
            raise exc

Context manager for a step inside a file.

Creates a :class:StepReport, times it, captures exceptions (no re-raise), and appends it to the provided :class:FileReport.

Parameters:

Name	Type	Description	Default
`file_report`	`FileReport`	Parent file report to append the step to.	required
`id`	`str`	Machine-friendly step id (e.g., `"extract_text"`).	required
`label`	`str`	Human-friendly label for UI display.	`None`

Yields:

Type	Description
`StepReport`	The live step to populate within the `with` block.

Notes

Exceptions are recorded on the step and status set to FAILED, but the exception is not re-raised (so the file can continue recording).
metadata['duration_ms'] is recorded on exit.

Source code in src/pipeline_watcher/core.py

@contextmanager
def file_step(file_report, id: str, *, label: str | None = None):
    """Context manager for a **step inside a file**.

    Creates a :class:`StepReport`, times it, captures exceptions (no re-raise),
    and appends it to the provided :class:`FileReport`.

    Parameters
    ----------
    file_report : FileReport
        Parent file report to append the step to.
    id : str
        Machine-friendly step id (e.g., ``"extract_text"``).
    label : str, optional
        Human-friendly label for UI display.

    Yields
    ------
    StepReport
        The live step to populate within the ``with`` block.

    Notes
    -----
    - Exceptions are recorded on the step and status set to ``FAILED``, but
      the exception is not re-raised (so the file can continue recording).
    - ``metadata['duration_ms']`` is recorded on exit.
    """
    st = StepReport.begin(id, label=label)
    t0 = time.perf_counter()
    try:
        yield st
        st.end()
    except BaseException as e:
        st.errors.append(f"{type(e).__name__}: {e}")
        st.metadata["traceback"] = traceback.format_exc()
        st.fail("Unhandled exception in file step")
        # By default, do not re-raise so the file continues recording
    finally:
        st.metadata["duration_ms"] = round((time.perf_counter() - t0) * 1000, 3)
        file_report.append_step(st)