from typing import TYPE_CHECKING

from sentry_sdk._batcher import Batcher

from sentry_sdk.utils import serialize_attribute

if TYPE_CHECKING:

from typing import Any

from sentry_sdk._types import Metric

class MetricsBatcher(Batcher["Metric"]):

MAX_BEFORE_FLUSH = 1000

MAX_BEFORE_DROP = 10_000

FLUSH_WAIT_TIME = 5.0

TYPE = "trace_metric"

CONTENT_TYPE = "application/vnd.sentry.items.trace-metric+json"

@staticmethod

def _to_transport_format(item: "Metric") -> "Any":

res = {

"timestamp": item["timestamp"],

"trace_id": item["trace_id"],

"name": item["name"],

"type": item["type"],

"value": item["value"],

"attributes": {

k: serialize_attribute(v) for (k, v) in item["attributes"].items()

},

}

if item.get("span_id") is not None:

res["span_id"] = item["span_id"]

if item.get("unit") is not None:

res["unit"] = item["unit"]

return res

def _record_lost(self, item: "Metric") -> None:

self._record_lost_func(

reason="queue_overflow",

data_category="trace_metric",

quantity=1,

)