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: envelope.py

import io
import json
import mimetypes
from typing import TYPE_CHECKING

from sentry_sdk.session import Session
from sentry_sdk.utils import capture_internal_exceptions, json_dumps

if TYPE_CHECKING:
    from typing import Any, Dict, Iterator, List, Optional, Union

from sentry_sdk._types import Event, EventDataCategory

def parse_json(data: "Union[bytes, str]") -> "Any":
    # on some python 3 versions this needs to be bytes
    if isinstance(data, bytes):
        data = data.decode("utf-8", "replace")
    return json.loads(data)

class Envelope:
    """
    Represents a Sentry Envelope. The calling code is responsible for adhering to the constraints
    documented in the Sentry docs: https://develop.sentry.dev/sdk/envelopes/#data-model. In particular,
    each envelope may have at most one Item with type "event" or "transaction" (but not both).
    """

def __init__(
        self,
        headers: "Optional[Dict[str, Any]]" = None,
        items: "Optional[List[Item]]" = None,
    ) -> None:
        if headers is not None:
            headers = dict(headers)
        self.headers = headers or {}
        if items is None:
            items = []
        else:
            items = list(items)
        self.items = items

@property
    def description(self) -> str:
        return "envelope with %s items (%s)" % (
            len(self.items),
            ", ".join(x.data_category for x in self.items),
        )

def add_event(
        self,
        event: "Event",
    ) -> None:
        self.add_item(Item(payload=PayloadRef(json=event), type="event"))

def add_transaction(
        self,
        transaction: "Event",
    ) -> None:
        self.add_item(Item(payload=PayloadRef(json=transaction), type="transaction"))

def add_profile(
        self,
        profile: "Any",
    ) -> None:
        self.add_item(Item(payload=PayloadRef(json=profile), type="profile"))

def add_profile_chunk(
        self,
        profile_chunk: "Any",
    ) -> None:
        self.add_item(
            Item(
                payload=PayloadRef(json=profile_chunk),
                type="profile_chunk",
                headers={"platform": profile_chunk.get("platform", "python")},
            )
        )

def add_checkin(
        self,
        checkin: "Any",
    ) -> None:
        self.add_item(Item(payload=PayloadRef(json=checkin), type="check_in"))

def add_session(
        self,
        session: "Union[Session, Any]",
    ) -> None:
        if isinstance(session, Session):
            session = session.to_json()
        self.add_item(Item(payload=PayloadRef(json=session), type="session"))

def add_sessions(
        self,
        sessions: "Any",
    ) -> None:
        self.add_item(Item(payload=PayloadRef(json=sessions), type="sessions"))

def add_item(
        self,
        item: "Item",
    ) -> None:
        self.items.append(item)

def get_event(self) -> "Optional[Event]":
        for items in self.items:
            event = items.get_event()
            if event is not None:
                return event
        return None

def get_transaction_event(self) -> "Optional[Event]":
        for item in self.items:
            event = item.get_transaction_event()
            if event is not None:
                return event
        return None

def __iter__(self) -> "Iterator[Item]":
        return iter(self.items)

def serialize_into(
        self,
        f: "Any",
    ) -> None:
        f.write(json_dumps(self.headers))
        f.write(b"\n")
        for item in self.items:
            item.serialize_into(f)

def serialize(self) -> bytes:
        out = io.BytesIO()
        self.serialize_into(out)
        return out.getvalue()

@classmethod
    def deserialize_from(
        cls,
        f: "Any",
    ) -> "Envelope":
        headers = parse_json(f.readline())
        items = []
        while 1:
            item = Item.deserialize_from(f)
            if item is None:
                break
            items.append(item)
        return cls(headers=headers, items=items)

@classmethod
    def deserialize(
        cls,
        bytes: bytes,
    ) -> "Envelope":
        return cls.deserialize_from(io.BytesIO(bytes))

def __repr__(self) -> str:
        return "<Envelope headers=%r items=%r>" % (self.headers, self.items)

class PayloadRef:
    def __init__(
        self,
        bytes: "Optional[bytes]" = None,
        path: "Optional[Union[bytes, str]]" = None,
        json: "Optional[Any]" = None,
    ) -> None:
        self.json = json
        self.bytes = bytes
        self.path = path

def get_bytes(self) -> bytes:
        if self.bytes is None:
            if self.path is not None:
                with capture_internal_exceptions():
                    with open(self.path, "rb") as f:
                        self.bytes = f.read()
            elif self.json is not None:
                self.bytes = json_dumps(self.json)
        return self.bytes or b""

@property
    def inferred_content_type(self) -> str:
        if self.json is not None:
            return "application/json"
        elif self.path is not None:
            path = self.path
            if isinstance(path, bytes):
                path = path.decode("utf-8", "replace")
            ty = mimetypes.guess_type(path)[0]
            if ty:
                return ty
        return "application/octet-stream"

def __repr__(self) -> str:
        return "<Payload %r>" % (self.inferred_content_type,)

class Item:
    def __init__(
        self,
        payload: "Union[bytes, str, PayloadRef]",
        headers: "Optional[Dict[str, Any]]" = None,
        type: "Optional[str]" = None,
        content_type: "Optional[str]" = None,
        filename: "Optional[str]" = None,
    ):
        if headers is not None:
            headers = dict(headers)
        elif headers is None:
            headers = {}
        self.headers = headers
        if isinstance(payload, bytes):
            payload = PayloadRef(bytes=payload)
        elif isinstance(payload, str):
            payload = PayloadRef(bytes=payload.encode("utf-8"))
        else:
            payload = payload

if filename is not None:
            headers["filename"] = filename
        if type is not None:
            headers["type"] = type
        if content_type is not None:
            headers["content_type"] = content_type
        elif "content_type" not in headers:
            headers["content_type"] = payload.inferred_content_type

self.payload = payload

def __repr__(self) -> str:
        return "<Item headers=%r payload=%r data_category=%r>" % (
            self.headers,
            self.payload,
            self.data_category,
        )

@property
    def type(self) -> "Optional[str]":
        return self.headers.get("type")

@property
    def data_category(self) -> "EventDataCategory":
        ty = self.headers.get("type")
        if ty == "session" or ty == "sessions":
            return "session"
        elif ty == "attachment":
            return "attachment"
        elif ty == "transaction":
            return "transaction"
        elif ty == "span":
            return "span"
        elif ty == "event":
            return "error"
        elif ty == "log":
            return "log_item"
        elif ty == "trace_metric":
            return "trace_metric"
        elif ty == "client_report":
            return "internal"
        elif ty == "profile":
            return "profile"
        elif ty == "profile_chunk":
            return "profile_chunk"
        elif ty == "check_in":
            return "monitor"
        else:
            return "default"

def get_bytes(self) -> bytes:
        return self.payload.get_bytes()

def get_event(self) -> "Optional[Event]":
        """
        Returns an error event if there is one.
        """
        if self.type == "event" and self.payload.json is not None:
            return self.payload.json
        return None

def get_transaction_event(self) -> "Optional[Event]":
        if self.type == "transaction" and self.payload.json is not None:
            return self.payload.json
        return None

def serialize_into(
        self,
        f: "Any",
    ) -> None:
        headers = dict(self.headers)
        bytes = self.get_bytes()
        headers["length"] = len(bytes)
        f.write(json_dumps(headers))
        f.write(b"\n")
        f.write(bytes)
        f.write(b"\n")

def serialize(self) -> bytes:
        out = io.BytesIO()
        self.serialize_into(out)
        return out.getvalue()

@classmethod
    def deserialize_from(
        cls,
        f: "Any",
    ) -> "Optional[Item]":
        line = f.readline().rstrip()
        if not line:
            return None
        headers = parse_json(line)
        length = headers.get("length")
        if length is not None:
            payload = f.read(length)
            f.readline()
        else:
            # if no length was specified we need to read up to the end of line
            # and remove it (if it is present, i.e. not the very last char in an eof terminated envelope)
            payload = f.readline().rstrip(b"\n")
        if headers.get("type") in ("event", "transaction"):
            rv = cls(headers=headers, payload=PayloadRef(json=parse_json(payload)))
        else:
            rv = cls(headers=headers, payload=payload)
        return rv

@classmethod
    def deserialize(
        cls,
        bytes: bytes,
    ) -> "Optional[Item]":
        return cls.deserialize_from(io.BytesIO(bytes))

X7ROOT File Manager

Editing: envelope.py

Upload File

Create Folder