Function record_exception

sdk/python/src/dagger/mod/_exceptions.py:127–161

Convert a Python exception into a `dagger.Error`.

(exc: Exception)

async def record_exception(exc: Exception):
    """Convert a Python exception into a `dagger.Error`."""
    attrs: dict[str, Any] = _exception_attributes(exc)
    msg = f"{attrs[EXCEPTION_TYPE]}: {attrs[EXCEPTION_MESSAGE]}"

    if isinstance(exc, ModuleError) and exc.extra:
        extra = {f"extra.{key}": val for key, val in exc.extra.items()}
        # ModuleError extra values don't conflict with the OTel attributes
        # but prepending like this avoids a future mistake.
        attrs = {**extra, **attrs}

    # Preserve original API error so it's properly propagated.
    if isinstance(exc, dagger.QueryError):
        msg = str(exc)
        attrs.update(exc.error.extensions)

    dag_err = dag.error(msg)
    for key, value in attrs.items():
        dag_err = dag_err.with_value(key, dagger.JSON(_safe_json_dumps(value)))

    await dag.current_function_call().return_error(dag_err)

    # When an error occurs within a started span context the OTel SDK
    # automatically sends an event with details about the exception.
    # Switching to dag.Error doesn't take advantage of that and the engine
    # doesn't recreate the exception event on the parent function span.
    # Still, recording the exception manually can be useful when analyzing the
    # raw telemetry in e.g., Honeycomb.
    with telemetry.get_tracer().start_as_current_span(
        "recording Python exception",
        # TODO: even with following attribute it's still being shown in the
        # Cloud UI.
        attributes={"dagger.io/ui.internal": True},
    ) as span:
        span.record_exception(exc)


def _exception_attributes(exc: Exception) -> dict[str, str]: