metric

requests = metric.counter("http_requests", attributes={"route": "/users"})
latency = metric.timer("http_latency")

handler = get_users().track_duration(latency).track(requests.with_constant_input(1))
text = yield from metric.prometheus_format()

boundaries_from_iterable(boundaries) -> tuple[float, ...]

def boundaries_from_iterable(boundaries: Iterable[float]) -> tuple[float, ...]:
    """The distinct positive ``boundaries``, in order, then ``math.inf``."""
    return (*(b for b in dict.fromkeys(boundaries) if b > 0), math.inf)

exponential_boundaries(*, start, factor, count) -> tuple[float, ...]

def exponential_boundaries(*, start: float, factor: float, count: int) -> tuple[float, ...]:
    """``count - 1`` boundaries (at least one) growing by ``factor`` from
    ``start``, then ``math.inf``."""
    return boundaries_from_iterable(start * _pow(factor, i) for i in range(max(1, count - 1)))

linear_boundaries(*, start, width, count) -> tuple[float, ...]

def linear_boundaries(*, start: float, width: float, count: int) -> tuple[float, ...]:
    """``count - 1`` boundaries (at least one) ``width`` apart from
    ``start``, then ``math.inf``."""
    return boundaries_from_iterable(start + n * width for n in range(max(1, count - 1)))

counter(name, *, description=None, attributes=None, bigint=False, incremental=False) -> Counter[Any]

def counter(
    name: str,
    *,
    description: str | None = None,
    attributes: Attributes | None = None,
    bigint: bool = False,
    incremental: bool = False,
) -> Counter[Any]:
    """A running total: ``update`` adds. An ``incremental`` counter ignores
    negative updates. ``bigint=True`` types it as ``int`` (Effect's
    ``bigint``); either way a counter of ints stays exact."""
    del bigint  # only the type differs: Python ints never lose precision
    return _Counter(name, description, attributes, incremental)

frequency(name, *, description=None, attributes=None, preregistered_words=()) -> Frequency

def frequency(
    name: str,
    *,
    description: str | None = None,
    attributes: Attributes | None = None,
    preregistered_words: Iterable[str] = (),
) -> Frequency:
    """How often each word is seen. ``preregistered_words`` start at 0."""
    return _Frequency(name, description, attributes, tuple(preregistered_words))

gauge(name, *, description=None, attributes=None, bigint=False) -> Gauge[Any]

def gauge(
    name: str,
    *,
    description: str | None = None,
    attributes: Attributes | None = None,
    bigint: bool = False,
) -> Gauge[Any]:
    """The latest value: ``update`` sets it, ``modify`` adds to it."""
    del bigint
    return _Gauge("Gauge", name, description, attributes)

histogram(name, *, boundaries, description=None, attributes=None) -> Histogram[float]

def histogram(
    name: str,
    *,
    boundaries: Iterable[float],
    description: str | None = None,
    attributes: Attributes | None = None,
) -> Histogram[float]:
    """Values counted into buckets. The state reports one cumulative
    bucket per boundary (sorted); values past the last are only in
    ``count`` (end the boundaries with ``math.inf`` to see them, as the
    ``*_boundaries`` helpers do)."""
    return _Histogram(name, description, attributes, tuple(sorted(map(float, boundaries))))

summary(name, *, max_age, max_size, quantiles, description=None, attributes=None) -> Summary[float]

def summary(
    name: str,
    *,
    max_age: Duration,
    max_size: int,
    quantiles: Sequence[float],
    description: str | None = None,
    attributes: Attributes | None = None,
) -> Summary[float]:
    """Quantiles over the last ``max_size`` values seen within ``max_age``
    (by the ``Clock``'s wall time), with the count, min, max and sum of
    every value. Each quantile is between 0 and 1."""
    inner = _summary(
        name,
        max_age=max_age,
        max_size=max_size,
        quantiles=quantiles,
        description=description,
        attributes=attributes,
    )
    return map_input_with_context(
        inner, lambda value, context: (value, context.unsafe_get(Clock).unsafe_now_nanos())
    )

summary_with_timestamp(name, *, max_age, max_size, quantiles, description=None, attributes=None) -> Summary[tuple[float, float]]

def summary_with_timestamp(
    name: str,
    *,
    max_age: Duration,
    max_size: int,
    quantiles: Sequence[float],
    description: str | None = None,
    attributes: Attributes | None = None,
) -> Summary[tuple[float, float]]:
    """``summary`` updated with ``(value, timestamp)``, the timestamp in
    wall-clock seconds since the epoch."""
    inner = _summary(
        name,
        max_age=max_age,
        max_size=max_size,
        quantiles=quantiles,
        description=description,
        attributes=attributes,
    )
    return map_input_with_context(inner, lambda observation, _: _stamped(*observation))

timer(name, *, description=None, attributes=None, boundaries=None) -> Histogram[Duration]

def timer(
    name: str,
    *,
    description: str | None = None,
    attributes: Attributes | None = None,
    boundaries: Iterable[float] | None = None,
) -> Histogram[Duration]:
    """A histogram of durations in milliseconds, with the attribute
    ``time_unit="milliseconds"``. The default boundaries double from 0.5ms,
    34 of them, then ``math.inf``."""
    bounds = (
        exponential_boundaries(start=0.5, factor=2, count=35) if boundaries is None else boundaries
    )
    with_unit = merge(attributes, {"time_unit": "milliseconds"})
    return histogram(
        name, boundaries=bounds, description=description, attributes=with_unit
    ).map_input(to_millis)

unsafe_register(metric, context, get) -> Registration

def unsafe_register[S](
    metric: Metric[Any, S], context: Context, get: Callable[[Context], S]
) -> Registration:
    """Make ``get`` the series ``context`` selects: snapshots and values
    call it, updates do nothing. ``remove()`` on the result drops the
    series, unless something registered it again since."""
    return metric._register(context, get)

prometheus_format(*, prefix=None, metric_name_mapper=None) -> Effect[str]

def prometheus_format(
    *, prefix: str | None = None, metric_name_mapper: MetricNameMapper | None = None
) -> Effect[str]:
    """The current registry as Prometheus text. Metric names go through
    ``metric_name_mapper``, are sanitized and get ``prefix_`` in front."""
    return WithContext(
        lambda context: unsafe_prometheus_format(
            context, prefix=prefix, metric_name_mapper=metric_name_mapper
        )
    )

unsafe_prometheus_format(context, *, prefix=None, metric_name_mapper=None) -> str

def unsafe_prometheus_format(
    context: Context,
    *,
    prefix: str | None = None,
    metric_name_mapper: MetricNameMapper | None = None,
) -> str:
    """``prometheus_format`` for ``context``'s registry."""
    start = _metric_name(prefix) + "_" if prefix else ""
    families: dict[str, list[Snapshot]] = {}
    for metric in unsafe_snapshot(context):
        mapped = metric.id if metric_name_mapper is None else metric_name_mapper(metric.id)
        families.setdefault(start + _metric_name(mapped), []).append(metric)
    lines: list[str] = []
    for name, metrics in families.items():
        first = metrics[0]
        if first.description:
            lines.append(f"# HELP {name} {_escape_help(first.description)}")
        lines.append(f"# TYPE {name} {_PROMETHEUS_TYPES[first.type]}")
        for metric in metrics:
            lines.extend(_samples(name, metric))
    return "\n".join(lines) + "\n" if lines else ""

disable_runtime_metrics(eff) -> Effect[A, E, R]

def disable_runtime_metrics[A, E, R](eff: Effect[A, E, R]) -> Effect[A, E, R]:
    """Run ``eff`` without recording fibers."""
    require_effect(eff, "eff")
    return eff.pipe(provide_service(FiberRuntimeMetrics, None))

enable_runtime_metrics(eff) -> Effect[A, E, R]

def enable_runtime_metrics[A, E, R](eff: Effect[A, E, R]) -> Effect[A, E, R]:
    """Run ``eff`` recording the fibers it starts in the runtime metrics."""
    require_effect(eff, "eff")
    return eff.pipe(provide_service(FiberRuntimeMetrics, fiber_runtime_metrics))

unsafe_snapshot(context) -> list[Snapshot]

def unsafe_snapshot(context: Context) -> list[Snapshot]:
    """Every series in ``context``'s registry, in registration order."""
    return [
        Snapshot(m.id, m.type, m.description, m.attributes, m.hooks.get(context))
        for m in list(context.unsafe_get(MetricRegistry).values())
    ]