FilesExpand file tree

master

/

session.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

174 lines (157 loc) · 5.41 KB

master

/

session.py

Copy path

Top

File metadata and controls

• Code
• Blame

174 lines (157 loc) · 5.41 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

166

167

168

169

170

171

172

173

174

import uuid

from datetime import datetime

from sentry_sdk._types import MYPY

from sentry_sdk.utils import format_timestamp

if MYPY:

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(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]

session_mode="application", # type: str

):

# 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.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.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 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