Tracing (Effect's Tracer): spans, the tracer that makes them, and the
references that steer them.
eff.with_span("name") runs eff inside a new span, the parent of
the spans made inside it; @fx.fn("name") does the same for every call
of a generator function::
@fx.fn("get_user")
def get_user(user_id: str):
yield from fx.annotate_current_span("user.id", user_id)
...
program.pipe(fx.provide(layer.span("startup")))The native tracer keeps spans in memory (NativeSpan) and exports
nothing; install another with eff.with_tracer(tracer) or a layer
providing CurrentTracer. Logs inside a span are recorded as its events
(logger.tracer_logger).
Spans also record where they were written: a failure inside with_span
or an @fx.fn call carries the StackFrame chain as its StackTrace
annotation, and Cause.pretty() prints it under the traceback. Turn it
off per span with capture_stack_trace=False.
Classes#
Functions#
annotate_current_span#
annotate_current_span(key_or_values, value=_MISSING) -> Effect[None]Set attributes on the current span, unless there is none or it has ended.
| Parameter | Type | Description |
|---|---|---|
key_or_values | str | Mapping[str, object] | |
value | object = _MISSING |
Returns Effect[None]
def annotate_current_span(
key_or_values: str | Mapping[str, object], value: object = _MISSING, /
) -> Effect[None]:
"""Set attributes on the current span, unless there is none or it has
ended."""
added = {key_or_values: value} if isinstance(key_or_values, str) else key_or_values
def body(fiber: Fiber[Any, Any]) -> Effect[None]:
span = local_span(fiber)
if span is not None and not isinstance(span.status, Ended):
for k, v in added.items():
span.attribute(k, v)
return _VOID
return _WithFiber(body)current_parent_span#
current_parent_span() -> Effect[AnySpan, NoSuchElementException]The current span, external or not; fails outside every span.
Returns Effect[AnySpan, NoSuchElementException]
def current_parent_span() -> Effect[AnySpan, NoSuchElementException]:
"""The current span, external or not; fails outside every span."""
def body(fiber: Fiber[Any, Any]) -> Effect[AnySpan, NoSuchElementException]:
span = fiber.get_ref(ParentSpan)
if span is None:
return _FailCause(fail_cause(NoSuchElementException()))
return _Succeed(span)
return _WithFiber(body)current_span#
current_span() -> Effect[Span, NoSuchElementException]The current span; fails when there is none or it is external.
Returns Effect[Span, NoSuchElementException]
def current_span() -> Effect[Span, NoSuchElementException]:
"""The current span; fails when there is none or it is external."""
def body(fiber: Fiber[Any, Any]) -> Effect[Span, NoSuchElementException]:
span = local_span(fiber)
if span is None:
return _FailCause(fail_cause(NoSuchElementException()))
return _Succeed(span)
return _WithFiber(body)make_span#
make_span(name, **options) -> Effect[Span]A new span, started now. Nothing ends it: end it yourself, or use
use_span / make_span_scoped.
| Parameter | Type | Description |
|---|---|---|
name | str | |
options | Unpack[SpanOptions] = {} |
Returns Effect[Span]
def make_span(name: str, **options: Unpack[SpanOptions]) -> Effect[Span]:
"""A new span, started now. Nothing ends it: end it yourself, or use
``use_span`` / ``make_span_scoped``."""
return _WithFiber(lambda fiber: _Succeed(make_span_unsafe(fiber, name, options)))make_span_scoped#
make_span_scoped(name, **options) -> Effect[Span, Never, Scope]A new span, ended when the current scope closes, with the scope's exit.
| Parameter | Type | Description |
|---|---|---|
name | str | |
options | Unpack[SpanOptions] = {} |
Returns Effect[Span, Never, Scope]
def make_span_scoped(name: str, **options: Unpack[SpanOptions]) -> Effect[Span, Never, Scope]:
"""A new span, ended when the current scope closes, with the scope's
exit."""
return span_scoped(name, options)use_span#
use_span(name, f, **options) -> Effect[A, E, R]Run f(span) with a new span, ended when the effect f returns
ends (or when f raises). The span is not made the parent of spans
inside; with_span does that.
| Parameter | Type | Description |
|---|---|---|
name | str | |
f | Callable[[Span], Effect[A, E, R]] | |
options | Unpack[SpanOptions] = {} |
Returns Effect[A, E, R]
def use_span[A, E = Never, R = Never](
name: str, f: Callable[[Span], Effect[A, E, R]], /, **options: Unpack[SpanOptions]
) -> Effect[A, E, R]:
"""Run ``f(span)`` with a new span, ended when the effect ``f`` returns
ends (or when ``f`` raises). The span is not made the parent of spans
inside; ``with_span`` does that."""
return _UseSpan(name, f, options)