Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/session.py: 11%
96 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1import uuid
2from datetime import datetime
4from sentry_sdk._types import MYPY
5from sentry_sdk.utils import format_timestamp
7if MYPY: 7 ↛ 8line 7 didn't jump to line 8, because the condition on line 7 was never true
8 from typing import Optional
9 from typing import Union
10 from typing import Any
11 from typing import Dict
13 from sentry_sdk._types import SessionStatus
16def _minute_trunc(ts):
17 # type: (datetime) -> datetime
18 return ts.replace(second=0, microsecond=0)
21def _make_uuid(
22 val, # type: Union[str, uuid.UUID]
23):
24 # type: (...) -> uuid.UUID
25 if isinstance(val, uuid.UUID):
26 return val
27 return uuid.UUID(val)
30class Session(object):
31 def __init__(
32 self,
33 sid=None, # type: Optional[Union[str, uuid.UUID]]
34 did=None, # type: Optional[str]
35 timestamp=None, # type: Optional[datetime]
36 started=None, # type: Optional[datetime]
37 duration=None, # type: Optional[float]
38 status=None, # type: Optional[SessionStatus]
39 release=None, # type: Optional[str]
40 environment=None, # type: Optional[str]
41 user_agent=None, # type: Optional[str]
42 ip_address=None, # type: Optional[str]
43 errors=None, # type: Optional[int]
44 user=None, # type: Optional[Any]
45 session_mode="application", # type: str
46 ):
47 # type: (...) -> None
48 if sid is None:
49 sid = uuid.uuid4()
50 if started is None:
51 started = datetime.utcnow()
52 if status is None:
53 status = "ok"
54 self.status = status
55 self.did = None # type: Optional[str]
56 self.started = started
57 self.release = None # type: Optional[str]
58 self.environment = None # type: Optional[str]
59 self.duration = None # type: Optional[float]
60 self.user_agent = None # type: Optional[str]
61 self.ip_address = None # type: Optional[str]
62 self.session_mode = session_mode # type: str
63 self.errors = 0
65 self.update(
66 sid=sid,
67 did=did,
68 timestamp=timestamp,
69 duration=duration,
70 release=release,
71 environment=environment,
72 user_agent=user_agent,
73 ip_address=ip_address,
74 errors=errors,
75 user=user,
76 )
78 @property
79 def truncated_started(self):
80 # type: (...) -> datetime
81 return _minute_trunc(self.started)
83 def update(
84 self,
85 sid=None, # type: Optional[Union[str, uuid.UUID]]
86 did=None, # type: Optional[str]
87 timestamp=None, # type: Optional[datetime]
88 started=None, # type: Optional[datetime]
89 duration=None, # type: Optional[float]
90 status=None, # type: Optional[SessionStatus]
91 release=None, # type: Optional[str]
92 environment=None, # type: Optional[str]
93 user_agent=None, # type: Optional[str]
94 ip_address=None, # type: Optional[str]
95 errors=None, # type: Optional[int]
96 user=None, # type: Optional[Any]
97 ):
98 # type: (...) -> None
99 # If a user is supplied we pull some data form it
100 if user:
101 if ip_address is None:
102 ip_address = user.get("ip_address")
103 if did is None:
104 did = user.get("id") or user.get("email") or user.get("username")
106 if sid is not None:
107 self.sid = _make_uuid(sid)
108 if did is not None:
109 self.did = str(did)
110 if timestamp is None:
111 timestamp = datetime.utcnow()
112 self.timestamp = timestamp
113 if started is not None:
114 self.started = started
115 if duration is not None:
116 self.duration = duration
117 if release is not None:
118 self.release = release
119 if environment is not None:
120 self.environment = environment
121 if ip_address is not None:
122 self.ip_address = ip_address
123 if user_agent is not None:
124 self.user_agent = user_agent
125 if errors is not None:
126 self.errors = errors
128 if status is not None:
129 self.status = status
131 def close(
132 self, status=None # type: Optional[SessionStatus]
133 ):
134 # type: (...) -> Any
135 if status is None and self.status == "ok":
136 status = "exited"
137 if status is not None:
138 self.update(status=status)
140 def get_json_attrs(
141 self, with_user_info=True # type: Optional[bool]
142 ):
143 # type: (...) -> Any
144 attrs = {}
145 if self.release is not None:
146 attrs["release"] = self.release
147 if self.environment is not None:
148 attrs["environment"] = self.environment
149 if with_user_info:
150 if self.ip_address is not None:
151 attrs["ip_address"] = self.ip_address
152 if self.user_agent is not None:
153 attrs["user_agent"] = self.user_agent
154 return attrs
156 def to_json(self):
157 # type: (...) -> Any
158 rv = {
159 "sid": str(self.sid),
160 "init": True,
161 "started": format_timestamp(self.started),
162 "timestamp": format_timestamp(self.timestamp),
163 "status": self.status,
164 } # type: Dict[str, Any]
165 if self.errors:
166 rv["errors"] = self.errors
167 if self.did is not None:
168 rv["did"] = self.did
169 if self.duration is not None:
170 rv["duration"] = self.duration
171 attrs = self.get_json_attrs()
172 if attrs:
173 rv["attrs"] = attrs
174 return rv