aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/sentry_sdk/session.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/sentry_sdk/session.py')
-rw-r--r--.venv/lib/python3.12/site-packages/sentry_sdk/session.py175
1 files changed, 175 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sentry_sdk/session.py b/.venv/lib/python3.12/site-packages/sentry_sdk/session.py
new file mode 100644
index 00000000..c1d422c1
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/sentry_sdk/session.py
@@ -0,0 +1,175 @@
+import uuid
+from datetime import datetime, timezone
+
+from sentry_sdk.utils import format_timestamp
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from typing import Optional
+ from typing import Union
+ from typing import Any
+ from typing import Dict
+
+ from sentry_sdk._types import SessionStatus
+
+
+def _minute_trunc(ts):
+ # type: (datetime) -> datetime
+ return ts.replace(second=0, microsecond=0)
+
+
+def _make_uuid(
+ val, # type: Union[str, uuid.UUID]
+):
+ # type: (...) -> uuid.UUID
+ if isinstance(val, uuid.UUID):
+ return val
+ return uuid.UUID(val)
+
+
+class Session:
+ 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]
+ session_mode="application", # type: str
+ ):
+ # type: (...) -> None
+ if sid is None:
+ sid = uuid.uuid4()
+ if started is None:
+ started = datetime.now(timezone.utc)
+ 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.session_mode = session_mode # type: 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,
+ )
+
+ @property
+ def truncated_started(self):
+ # type: (...) -> datetime
+ return _minute_trunc(self.started)
+
+ 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.now(timezone.utc)
+ 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 get_json_attrs(
+ self, with_user_info=True # type: Optional[bool]
+ ):
+ # type: (...) -> Any
+ attrs = {}
+ if self.release is not None:
+ attrs["release"] = self.release
+ if self.environment is not None:
+ attrs["environment"] = self.environment
+ if with_user_info:
+ 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
+ return attrs
+
+ 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 = self.get_json_attrs()
+ if attrs:
+ rv["attrs"] = attrs
+ return rv