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

import copy
from threading import Lock
from typing import TYPE_CHECKING, Any

import sentry_sdk
from sentry_sdk._lru_cache import LRUCache
from sentry_sdk.tracing import Span
from sentry_sdk.tracing_utils import has_span_streaming_enabled

if TYPE_CHECKING:
    from typing import TypedDict

FlagData = TypedDict("FlagData", {"flag": str, "result": bool})

DEFAULT_FLAG_CAPACITY = 100

class FlagBuffer:
    def __init__(self, capacity: int) -> None:
        self.capacity = capacity
        self.lock = Lock()

# Buffer is private. The name is mangled to discourage use. If you use this attribute
        # directly you're on your own!
        self.__buffer = LRUCache(capacity)

def clear(self) -> None:
        self.__buffer = LRUCache(self.capacity)

def __deepcopy__(self, memo: "dict[int, Any]") -> "FlagBuffer":
        with self.lock:
            buffer = FlagBuffer(self.capacity)
            buffer.__buffer = copy.deepcopy(self.__buffer, memo)
            return buffer

def get(self) -> "list[FlagData]":
        with self.lock:
            return [
                {"flag": key, "result": value} for key, value in self.__buffer.get_all()
            ]

def set(self, flag: str, result: bool) -> None:
        if isinstance(result, FlagBuffer):
            # If someone were to insert `self` into `self` this would create a circular dependency
            # on the lock. This is of course a deadlock. However, this is far outside the expected
            # usage of this class. We guard against it here for completeness and to document this
            # expected failure mode.
            raise ValueError(
                "FlagBuffer instances can not be inserted into the dictionary."
            )

with self.lock:
            self.__buffer.set(flag, result)

def add_feature_flag(flag: str, result: bool) -> None:
    """
    Records a flag and its value to be sent on subsequent error events.
    We recommend you do this on flag evaluations. Flags are buffered per Sentry scope.
    """
    client = sentry_sdk.get_client()

flags = sentry_sdk.get_isolation_scope().flags
    flags.set(flag, result)

if has_span_streaming_enabled(client.options):
        span = sentry_sdk.traces.get_current_span()
        if span and isinstance(span, sentry_sdk.traces.StreamedSpan):
            span.set_attribute(f"flag.evaluation.{flag}", result)

else:
        span = sentry_sdk.get_current_span()
        if span and isinstance(span, Span):
            span.set_flag(f"flag.evaluation.{flag}", result)

X7ROOT File Manager

Editing: feature_flags.py

Upload File

Create Folder