Repository navigation
Expand file tree
/
Copy pathsession.py
More file actions
165 lines (148 loc) · 5.08 KB
/
Copy pathsession.py
File metadata and controls
165 lines (148 loc) · 5.08 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from sentry_sdk.utils import format_timestamp
if TYPE_CHECKING:
from typing import Any, Dict, Optional, Union
from sentry_sdk._types import SessionStatus
def _minute_trunc(ts: "datetime") -> "datetime":
return ts.replace(second=0, microsecond=0)
def _make_uuid(
val: "Union[str, uuid.UUID]",
) -> "uuid.UUID":
if isinstance(val, uuid.UUID):
return val
return uuid.UUID(val)
class Session:
def __init__(
self,
sid: "Optional[Union[str, uuid.UUID]]" = None,
did: "Optional[str]" = None,
timestamp: "Optional[datetime]" = None,
started: "Optional[datetime]" = None,
duration: "Optional[float]" = None,
status: "Optional[SessionStatus]" = None,
release: "Optional[str]" = None,
environment: "Optional[str]" = None,
user_agent: "Optional[str]" = None,
ip_address: "Optional[str]" = None,
errors: "Optional[int]" = None,
user: "Optional[Any]" = None,
session_mode: str = "application",
) -> 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: "Optional[str]" = None
self.started = started
self.release: "Optional[str]" = None
self.environment: "Optional[str]" = None
self.duration: "Optional[float]" = None
self.user_agent: "Optional[str]" = None
self.ip_address: "Optional[str]" = None
self.session_mode: str = session_mode
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) -> "datetime":
return _minute_trunc(self.started)
def update(
self,
sid: "Optional[Union[str, uuid.UUID]]" = None,
did: "Optional[str]" = None,
timestamp: "Optional[datetime]" = None,
started: "Optional[datetime]" = None,
duration: "Optional[float]" = None,
status: "Optional[SessionStatus]" = None,
release: "Optional[str]" = None,
environment: "Optional[str]" = None,
user_agent: "Optional[str]" = None,
ip_address: "Optional[str]" = None,
errors: "Optional[int]" = None,
user: "Optional[Any]" = None,
) -> 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: "Optional[SessionStatus]" = None,
) -> "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: "Optional[bool]" = True,
) -> "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) -> "Any":
rv: "Dict[str, Any]" = {
"sid": str(self.sid),
"init": True,
"started": format_timestamp(self.started),
"timestamp": format_timestamp(self.timestamp),
"status": self.status,
}
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