iframe-proxy | Sunbelt Computer Software

History

251 lines (196 loc) · 7.65 KB

Raw

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

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

import functools

from typing import TYPE_CHECKING

import sentry_sdk

from sentry_sdk.consts import SPANDATA

from sentry_sdk.integrations import (

_DEFAULT_FAILED_REQUEST_STATUS_CODES,

DidNotEnable,

Integration,

_check_minimum_version,

)

from sentry_sdk.integrations._wsgi_common import RequestExtractor

from sentry_sdk.integrations.wsgi import SentryWsgiMiddleware

from sentry_sdk.traces import SOURCE_FOR_STYLE as SEGMENT_SOURCE_FOR_STYLE

from sentry_sdk.tracing import SOURCE_FOR_STYLE as TRANSACTION_SOURCE_FOR_STYLE

from sentry_sdk.tracing_utils import has_span_streaming_enabled

from sentry_sdk.utils import (

capture_internal_exceptions,

ensure_integration_enabled,

event_from_exception,

parse_version,

transaction_from_function,

)

if TYPE_CHECKING:

from collections.abc import Set

from typing import Any, Callable, Dict, Optional

from bottle import FileUpload, FormsDict, LocalRequest # type: ignore

from sentry_sdk._types import Event, EventProcessor

from sentry_sdk.integrations.wsgi import _ScopedResponse

try:

from bottle import (

Bottle,

HTTPResponse,

Route,

)

from bottle import (

__version__ as BOTTLE_VERSION,

)

from bottle import (

request as bottle_request,

)

except ImportError:

raise DidNotEnable("Bottle not installed")

TRANSACTION_STYLE_VALUES = ("endpoint", "url")

class BottleIntegration(Integration):

identifier = "bottle"

origin = f"auto.http.{identifier}"

transaction_style = ""

def __init__(

self,

transaction_style: str = "endpoint",

failed_request_status_codes: "Set[int]" = _DEFAULT_FAILED_REQUEST_STATUS_CODES,

) -> None:

if transaction_style not in TRANSACTION_STYLE_VALUES:

raise ValueError(

"Invalid value for transaction_style: %s (must be in %s)"

% (transaction_style, TRANSACTION_STYLE_VALUES)

)

self.transaction_style = transaction_style

self.failed_request_status_codes = failed_request_status_codes

@staticmethod

def setup_once() -> None:

version = parse_version(BOTTLE_VERSION)

_check_minimum_version(BottleIntegration, version)

old_app = Bottle.__call__

@ensure_integration_enabled(BottleIntegration, old_app)

def sentry_patched_wsgi_app(

self: "Any", environ: "Dict[str, str]", start_response: "Callable[..., Any]"

) -> "_ScopedResponse":

middleware = SentryWsgiMiddleware(

lambda *a, **kw: old_app(self, *a, **kw),

span_origin=BottleIntegration.origin,

)

return middleware(environ, start_response)

Bottle.__call__ = sentry_patched_wsgi_app

old_handle = Bottle._handle

@functools.wraps(old_handle)

def _patched_handle(self: "Bottle", environ: "Dict[str, Any]") -> "Any":

integration = sentry_sdk.get_client().get_integration(BottleIntegration)

if integration is None:

return old_handle(self, environ)

scope = sentry_sdk.get_isolation_scope()

scope._name = "bottle"

scope.add_event_processor(

_make_request_event_processor(self, bottle_request, integration)

)

res = old_handle(self, environ)

server_span = sentry_sdk.get_current_scope()._server_segment_span

if server_span is not None:

route_path = None

try:

route_path = bottle_request.route.rule

except RuntimeError:

pass

if route_path is not None:

server_span.set_attribute(SPANDATA.HTTP_ROUTE, route_path)

if has_span_streaming_enabled(sentry_sdk.get_client().options):

_set_segment_name_and_source(

transaction_style=integration.transaction_style

)

return res

Bottle._handle = _patched_handle

old_make_callback = Route._make_callback

@functools.wraps(old_make_callback)

def patched_make_callback(

self: "Route", *args: object, **kwargs: object

) -> "Any":

prepared_callback = old_make_callback(self, *args, **kwargs)

integration = sentry_sdk.get_client().get_integration(BottleIntegration)

if integration is None:

return prepared_callback

def wrapped_callback(*args: object, **kwargs: object) -> "Any":

try:

res = prepared_callback(*args, **kwargs)

except Exception as exception:

_capture_exception(exception, handled=False)

raise exception

if (

isinstance(res, HTTPResponse)

and res.status_code in integration.failed_request_status_codes

_capture_exception(res, handled=True)

return res

return wrapped_callback

Route._make_callback = patched_make_callback

class BottleRequestExtractor(RequestExtractor):

def env(self) -> "Dict[str, str]":

return self.request.environ

def cookies(self) -> "Dict[str, str]":

return self.request.cookies

def raw_data(self) -> bytes:

return self.request.body.read()

def form(self) -> "FormsDict":

if self.is_json():

return None

return self.request.forms.decode()

def files(self) -> "Optional[Dict[str, str]]":

if self.is_json():

return None

return self.request.files

def size_of_file(self, file: "FileUpload") -> int:

return file.content_length

def _set_segment_name_and_source(transaction_style: str) -> None:

try:

if transaction_style == "url":

name = bottle_request.route.rule or "bottle request"

else:

name = (

bottle_request.route.name

or transaction_from_function(bottle_request.route.callback)

or "bottle request"

)

sentry_sdk.get_current_scope().set_transaction_name(

name,

source=SEGMENT_SOURCE_FOR_STYLE[transaction_style],

)

except RuntimeError:

pass

def _set_transaction_name_and_source(

event: "Event", transaction_style: str, request: "Any"

) -> None:

name = ""

if transaction_style == "url":

try:

name = request.route.rule or ""

except RuntimeError:

pass

elif transaction_style == "endpoint":

try:

name = (

request.route.name

or transaction_from_function(request.route.callback)

or ""

)

except RuntimeError:

pass

event["transaction"] = name

event["transaction_info"] = {

"source": TRANSACTION_SOURCE_FOR_STYLE[transaction_style]

}

def _make_request_event_processor(

app: "Bottle", request: "LocalRequest", integration: "BottleIntegration"

) -> "EventProcessor":

def event_processor(event: "Event", hint: "dict[str, Any]") -> "Event":

_set_transaction_name_and_source(event, integration.transaction_style, request)

with capture_internal_exceptions():

BottleRequestExtractor(request).extract_into_event(event)

return event

return event_processor

def _capture_exception(exception: BaseException, handled: bool) -> None:

event, hint = event_from_exception(

exception,

client_options=sentry_sdk.get_client().options,

mechanism={"type": "bottle", "handled": handled},

)

sentry_sdk.capture_event(event, hint=hint)

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

bottle.py

Latest commit

History

bottle.py

File metadata and controls