Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/session.py: 11%

1import uuid 

2from datetime import datetime 

3 

4from sentry_sdk._types import MYPY 

5from sentry_sdk.utils import format_timestamp 

6 

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 

12 

13 from sentry_sdk._types import SessionStatus 

14 

15 

16def _minute_trunc(ts): 

17 # type: (datetime) -> datetime 

18 return ts.replace(second=0, microsecond=0) 

19 

20 

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) 

28 

29 

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 

64 

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 ) 

77 

78 @property 

79 def truncated_started(self): 

80 # type: (...) -> datetime 

81 return _minute_trunc(self.started) 

82 

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") 

105 

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 

127 

128 if status is not None: 

129 self.status = status 

130 

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) 

139 

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 

155 

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