File Manager

X7ROOT File Manager

Current Path: /opt/hc_python/lib/python3.12/site-packages/sentry_sdk

opt / hc_python / lib / python3.12 / site-packages / sentry_sdk /

?? ..
?? __init__.py(1.46 KB)
?? __pycache__
?? _batcher.py(5.7 KB)
?? _compat.py(3 KB)
?? _init_implementation.py(2.43 KB)
?? _log_batcher.py(1.88 KB)
?? _lru_cache.py(1.14 KB)
?? _metrics_batcher.py(1.21 KB)
?? _queue.py(10.98 KB)
?? _span_batcher.py(8.12 KB)
?? _types.py(13.16 KB)
?? _werkzeug.py(3.85 KB)
?? ai
?? api.py(15.59 KB)
?? attachments.py(2.95 KB)
?? client.py(49.95 KB)
?? consts.py(61.95 KB)
?? crons
?? debug.py(959 B)
?? envelope.py(9.37 KB)
?? feature_flags.py(2.5 KB)
?? hub.py(24.54 KB)
?? integrations
?? logger.py(2.6 KB)
?? metrics.py(1.42 KB)
?? monitor.py(4.47 KB)
?? profiler
?? py.typed(0 B)
?? scope.py(74.09 KB)
?? scrubber.py(5.99 KB)
?? serializer.py(12.82 KB)
?? session.py(5.08 KB)
?? sessions.py(8.59 KB)
?? spotlight.py(11.85 KB)
?? traces.py(25.08 KB)
?? tracing.py(50.33 KB)
?? tracing_utils.py(54.36 KB)
?? transport.py(44.41 KB)
?? types.py(1.24 KB)
?? utils.py(65.96 KB)
?? worker.py(10.91 KB)

Editing: _log_batcher.py

from typing import TYPE_CHECKING

from sentry_sdk._batcher import Batcher
from sentry_sdk.envelope import Item, PayloadRef
from sentry_sdk.utils import serialize_attribute

if TYPE_CHECKING:
    from typing import Any

from sentry_sdk._types import Log

class LogBatcher(Batcher["Log"]):
    MAX_BEFORE_FLUSH = 100
    MAX_BEFORE_DROP = 1_000
    FLUSH_WAIT_TIME = 5.0

TYPE = "log"
    CONTENT_TYPE = "application/vnd.sentry.items.log+json"

@staticmethod
    def _to_transport_format(item: "Log") -> "Any":
        if "sentry.severity_number" not in item["attributes"]:
            item["attributes"]["sentry.severity_number"] = item["severity_number"]
        if "sentry.severity_text" not in item["attributes"]:
            item["attributes"]["sentry.severity_text"] = item["severity_text"]

res = {
            "timestamp": int(item["time_unix_nano"]) / 1.0e9,
            "trace_id": item.get("trace_id", "00000000-0000-0000-0000-000000000000"),
            "span_id": item.get("span_id"),
            "level": str(item["severity_text"]),
            "body": str(item["body"]),
            "attributes": {
                k: serialize_attribute(v) for (k, v) in item["attributes"].items()
            },
        }

return res

def _record_lost(self, item: "Log") -> None:
        # Construct log envelope item without sending it to report lost bytes
        log_item = Item(
            type=self.TYPE,
            content_type=self.CONTENT_TYPE,
            headers={
                "item_count": 1,
            },
            payload=PayloadRef(
                json={
                    "version": 2,
                    "items": [self._to_transport_format(item)],
                }
            ),
        )

self._record_lost_func(
            reason="queue_overflow",
            data_category="log_item",
            item=log_item,
            quantity=1,
        )

X7ROOT File Manager

Editing: _log_batcher.py

Upload File

Create Folder