FilesExpand file tree

master

/

session.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

165 lines (148 loc) · 5.08 KB

master

/

session.py

Copy path

Top

File metadata and controls

• Code
• Blame

165 lines (148 loc) · 5.08 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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