ACIL FM

Nama	Ukuran	Aksi
integrations	-	Open
__pycache__	-	Open
api.py	4.69 MB	View DL Edit
attachments.py	1.75 MB	View DL Edit
client.py	14.03 MB	View DL Edit
consts.py	3.5 MB	View DL Edit
debug.py	1.11 MB	View DL Edit
envelope.py	8.17 MB	View DL Edit
hub.py	21.35 MB	View DL Edit
py.typed	0 B	View DL Edit
scope.py	15.62 MB	View DL Edit
serializer.py	15.92 MB	View DL Edit
sessions.py	7.67 MB	View DL Edit
tracing.py	25.24 MB	View DL Edit
transport.py	11.48 MB	View DL Edit
utils.py	26.26 MB	View DL Edit
worker.py	3.87 MB	View DL Edit
_compat.py	2.3 MB	View DL Edit
_functools.py	2.22 MB	View DL Edit
_queue.py	8.21 MB	View DL Edit
_types.py	1.26 MB	View DL Edit
__init__.py	854 B	View DL Edit

Edit file: /opt/imunify360/venv/lib/python3.11/site-packages/sentry_sdk/sessions.py

import os
import uuid
import time
from datetime import datetime
from threading import Thread, Lock
from contextlib import contextmanager

from sentry_sdk._types import MYPY
from sentry_sdk.utils import format_timestamp

if MYPY:
    import sentry_sdk

from typing import Optional
    from typing import Union
    from typing import Any
    from typing import Dict
    from typing import Generator

from sentry_sdk._types import SessionStatus

def is_auto_session_tracking_enabled(hub=None):
    # type: (Optional[sentry_sdk.Hub]) -> bool
    """Utility function to find out if session tracking is enabled."""
    if hub is None:
        hub = sentry_sdk.Hub.current
    should_track = hub.scope._force_auto_session_tracking
    if should_track is None:
        exp = hub.client.options["_experiments"] if hub.client else {}
        should_track = exp.get("auto_session_tracking")
    return should_track

@contextmanager
def auto_session_tracking(hub=None):
    # type: (Optional[sentry_sdk.Hub]) -> Generator[None, None, None]
    """Starts and stops a session automatically around a block."""
    if hub is None:
        hub = sentry_sdk.Hub.current
    should_track = is_auto_session_tracking_enabled(hub)
    if should_track:
        hub.start_session()
    try:
        yield
    finally:
        if should_track:
            hub.end_session()

def _make_uuid(
    val,  # type: Union[str, uuid.UUID]
):
    # type: (...) -> uuid.UUID
    if isinstance(val, uuid.UUID):
        return val
    return uuid.UUID(val)

TERMINAL_SESSION_STATES = ("exited", "abnormal", "crashed")

class SessionFlusher(object):
    def __init__(
        self,
        flush_func,  # type: Any
        flush_interval=10,  # type: int
    ):
        # type: (...) -> None
        self.flush_func = flush_func
        self.flush_interval = flush_interval
        self.pending = {}  # type: Dict[str, Any]
        self._thread = None  # type: Optional[Thread]
        self._thread_lock = Lock()
        self._thread_for_pid = None  # type: Optional[int]
        self._running = True

def flush(self):
        # type: (...) -> None
        pending = self.pending
        self.pending = {}
        self.flush_func(list(pending.values()))

def _ensure_running(self):
        # type: (...) -> None
        if self._thread_for_pid == os.getpid() and self._thread is not None:
            return None
        with self._thread_lock:
            if self._thread_for_pid == os.getpid() and self._thread is not None:
                return None

def _thread():
                # type: (...) -> None
                while self._running:
                    time.sleep(self.flush_interval)
                    if self.pending and self._running:
                        self.flush()

thread = Thread(target=_thread)
            thread.daemon = True
            thread.start()
            self._thread = thread
            self._thread_for_pid = os.getpid()
        return None

def add_session(
        self, session  # type: Session
    ):
        # type: (...) -> None
        self.pending[session.sid.hex] = session.to_json()
        self._ensure_running()

def kill(self):
        # type: (...) -> None
        self._running = False

def __del__(self):
        # type: (...) -> None
        self.kill()

class Session(object):
    def __init__(
        self,
        sid=None,  # type: Optional[Union[str, uuid.UUID]]
        did=None,  # type: Optional[str]
        timestamp=None,  # type: Optional[datetime]
        started=None,  # type: Optional[datetime]
        duration=None,  # type: Optional[float]
        status=None,  # type: Optional[SessionStatus]
        release=None,  # type: Optional[str]
        environment=None,  # type: Optional[str]
        user_agent=None,  # type: Optional[str]
        ip_address=None,  # type: Optional[str]
        errors=None,  # type: Optional[int]
        user=None,  # type: Optional[Any]
    ):
        # type: (...) -> None
        if sid is None:
            sid = uuid.uuid4()
        if started is None:
            started = datetime.utcnow()
        if status is None:
            status = "ok"
        self.status = status
        self.did = None  # type: Optional[str]
        self.started = started
        self.release = None  # type: Optional[str]
        self.environment = None  # type: Optional[str]
        self.duration = None  # type: Optional[float]
        self.user_agent = None  # type: Optional[str]
        self.ip_address = None  # type: Optional[str]
        self.errors = 0

self.update(
            sid=sid,
            did=did,
            timestamp=timestamp,
            duration=duration,
            release=release,
            environment=environment,
            user_agent=user_agent,
            ip_address=ip_address,
            errors=errors,
            user=user,
        )

def update(
        self,
        sid=None,  # type: Optional[Union[str, uuid.UUID]]
        did=None,  # type: Optional[str]
        timestamp=None,  # type: Optional[datetime]
        started=None,  # type: Optional[datetime]
        duration=None,  # type: Optional[float]
        status=None,  # type: Optional[SessionStatus]
        release=None,  # type: Optional[str]
        environment=None,  # type: Optional[str]
        user_agent=None,  # type: Optional[str]
        ip_address=None,  # type: Optional[str]
        errors=None,  # type: Optional[int]
        user=None,  # type: Optional[Any]
    ):
        # type: (...) -> None
        # If a user is supplied we pull some data form it
        if user:
            if ip_address is None:
                ip_address = user.get("ip_address")
            if did is None:
                did = user.get("id") or user.get("email") or user.get("username")

if sid is not None:
            self.sid = _make_uuid(sid)
        if did is not None:
            self.did = str(did)
        if timestamp is None:
            timestamp = datetime.utcnow()
        self.timestamp = timestamp
        if started is not None:
            self.started = started
        if duration is not None:
            self.duration = duration
        if release is not None:
            self.release = release
        if environment is not None:
            self.environment = environment
        if ip_address is not None:
            self.ip_address = ip_address
        if user_agent is not None:
            self.user_agent = user_agent
        if errors is not None:
            self.errors = errors

if status is not None:
            self.status = status

def close(
        self, status=None  # type: Optional[SessionStatus]
    ):
        # type: (...) -> Any
        if status is None and self.status == "ok":
            status = "exited"
        if status is not None:
            self.update(status=status)

def to_json(self):
        # type: (...) -> Any
        rv = {
            "sid": str(self.sid),
            "init": True,
            "started": format_timestamp(self.started),
            "timestamp": format_timestamp(self.timestamp),
            "status": self.status,
        }  # type: Dict[str, Any]
        if self.errors:
            rv["errors"] = self.errors
        if self.did is not None:
            rv["did"] = self.did
        if self.duration is not None:
            rv["duration"] = self.duration

attrs = {}
        if self.release is not None:
            attrs["release"] = self.release
        if self.environment is not None:
            attrs["environment"] = self.environment
        if self.ip_address is not None:
            attrs["ip_address"] = self.ip_address
        if self.user_agent is not None:
            attrs["user_agent"] = self.user_agent
        if attrs:
            rv["attrs"] = attrs
        return rv

Batal