Events & Reporting

Classes and utilities for reporting job progress and events.

events

EventContext

Source code in ivcap_service/events.py

class EventContext:
    def __init__(
        self,
        event_name: str,
        reporter: "EventReporter",
        finishEventClass: type[GenericEvent] | None,
        errorEventClass: type[GenericErrorEvent] | None,
        span_attributes: dict[str, Any] | None = None,
    ):
        self._event_name = event_name
        self._reporter = reporter
        self._finishEventClass = finishEventClass
        self._errorEventClass = errorEventClass
        self._finished_sent = False

        # Best-effort OpenTelemetry child span for the event scope.
        # This should never break services that don't use OTEL.
        self._otel_span_cm = None
        self._otel_span = None
        try:
            from opentelemetry import trace

            tracer = trace.get_tracer("ivcap_service.events")
            self._otel_span_cm = tracer.start_as_current_span(
                f"ivcap.event:{event_name}"
            )
            self._otel_span = self._otel_span_cm.__enter__()
            try:
                self._otel_span.set_attribute("ivcap.job_id", reporter.job_id)
                self._otel_span.set_attribute("ivcap.event_name", event_name)
                service_id = os.getenv("IVCAP_SERVICE_ID")
                if service_id:
                    self._otel_span.set_attribute("ivcap.service_id", service_id)
                # Extra caller-supplied kwargs (e.g. from `report.step(..., foo=1)`)
                # are set as `ivcap.event.<key>` attributes, best-effort. Values
                # that OTel can't represent (dicts, None, mixed-type lists, ...)
                # are silently skipped rather than raising.
                for k, v in (span_attributes or {}).items():
                    coerced = _otel_attribute_value(v)
                    if coerced is not None:
                        self._otel_span.set_attribute(f"ivcap.event.{k}", coerced)
            except Exception:
                pass
        except Exception:
            pass

    @property
    def name(self):
        return self._event_name

    def finished(self, message=None, **kwargs):
        options = kwargs
        if message:
            options["message"] = message
        if self._finishEventClass:
            event = self._finishEventClass(name=self._event_name, options=options)
        else:
            event = GenericEvent(name=self._event_name, options=options)
        self._reporter.emit(event)
        self._finished_sent = True

        if self._otel_span is not None:
            try:
                for k, v in options.items():
                    coerced = _otel_attribute_value(v)
                    if coerced is not None:
                        self._otel_span.set_attribute(f"ivcap.event.{k}", coerced)
            except Exception:
                pass
            try:
                from opentelemetry.trace import Status, StatusCode

                self._otel_span.set_status(Status(StatusCode.OK))
            except Exception:
                pass

    def info(self, event: BaseEvent | dict):
        # We keep this ergonomic for callers: if they pass a raw dict,
        # wrap it into a generic event.
        if isinstance(event, BaseEvent):
            self._reporter.emit(event)
        else:
            self._reporter.emit(GenericEvent(name=self._event_name, options=event))

    def error(self, err: Exception, context: str | None = None):
        evc = (
            self._errorEventClass
            if self._errorEventClass is not None
            else GenericErrorEvent
        )
        stacktrace = traceback.format_tb(err.__traceback__)
        if not context:
            context = self._event_name
        event = evc(error=str(err), stacktrace=stacktrace, context=context)
        self._reporter.emit(event)

        if self._otel_span is not None:
            try:
                self._otel_span.record_exception(err)
            except Exception:
                pass
            try:
                from opentelemetry.trace import Status, StatusCode

                self._otel_span.set_status(
                    Status(StatusCode.ERROR, description=str(err))
                )
            except Exception:
                pass

    def close(self):
        """Close the underlying OTEL span if one was created."""

        if self._otel_span_cm is not None:
            try:
                self._otel_span_cm.__exit__(None, None, None)
            except Exception:
                pass
            finally:
                self._otel_span_cm = None
                self._otel_span = None

    def __del__(self):
        # Best-effort cleanup in case callers forget to exit the context.
        try:
            self.close()
        except Exception:
            pass

close()

Close the underlying OTEL span if one was created.

Source code in ivcap_service/events.py

def close(self):
    """Close the underlying OTEL span if one was created."""

    if self._otel_span_cm is not None:
        try:
            self._otel_span_cm.__exit__(None, None, None)
        except Exception:
            pass
        finally:
            self._otel_span_cm = None
            self._otel_span = None

set_event_reporter_factory(factory)

Set a factory function for creating specialised EventReporter instances.

Parameters:

Name	Type	Description	Default
factory	EventFactoryF | None	A factory function that takes a job ID and an optional job authorization token,	required

Source code in ivcap_service/events.py

def set_event_reporter_factory(factory: EventFactoryF | None):
    """
    Set a factory function for creating specialised EventReporter instances.

    Args:
        factory (EventFactoryF | None): A factory function that takes a job ID and an optional job authorization token,
        and returns an instance of EventReporter. If None, the default EventReporter will be used.
    """
    global event_reporter_factory
    if factory is not None and not callable(factory):
        raise ValueError(
            "Factory must be a callable that returns an EventReporter instance."
        )
    event_reporter_factory = factory

create_event_reporter(job_id, job_authorization=None)

Create an EventReporter instance for the given job ID.

Parameters:

Name	Type	Description	Default
job_id	str	The unique identifier for the job.	required
job_authorization	Optional[str]	Optional authorization token for the job.	None

Returns:

Name	Type	Description
EventReporter	EventReporter	An instance of EventReporter initialized with the job ID.

Source code in ivcap_service/events.py

def create_event_reporter(
    job_id: str, job_authorization: str | None = None
) -> "EventReporter":
    """
    Create an EventReporter instance for the given job ID.


    Args:
        job_id (str): The unique identifier for the job.
        job_authorization (Optional[str]): Optional authorization token for the job.

    Returns:
        EventReporter: An instance of EventReporter initialized with the job ID.
    """
    if event_reporter_factory is not None:
        return event_reporter_factory(job_id, job_authorization)
    return EventReporter(job_id, job_authorization)

Copy Markdown