The method is invoked on every request and shows the lifecycle of the request received from the middleware. Although some middleware may use parts of the API spec, it is safe to assume that if you're looking for the particular spec property handler, it's most likely
(self, req)
| 319 | return endpoint, path_vars |
| 320 | |
| 321 | def __call__(self, req): |
| 322 | """ |
| 323 | The method is invoked on every request and shows the lifecycle of the request received from |
| 324 | the middleware. |
| 325 | |
| 326 | Although some middleware may use parts of the API spec, it is safe to assume that if you're |
| 327 | looking for the particular spec property handler, it's most likely a part of this method. |
| 328 | |
| 329 | At the time of writing, the only property being utilized by middleware was `x-log-result`. |
| 330 | """ |
| 331 | LOG.debug("Received call with WebOb: %s %s", req.method, req.url) |
| 332 | # if a more detailed log is required: |
| 333 | # loggable_req = req.copy() |
| 334 | # loggable_req.headers.pop('Authorization', None) |
| 335 | # loggable_req.headers.pop('X-Request-Id', None) |
| 336 | # LOG.debug("Received call with WebOb: %s", loggable_req) |
| 337 | endpoint, path_vars = self.match(req) |
| 338 | LOG.debug("Parsed endpoint: %s", endpoint) |
| 339 | LOG.debug("Parsed path_vars: %s", path_vars) |
| 340 | |
| 341 | context = copy.copy(getattr(self, "mock_context", {})) |
| 342 | cookie_token = None |
| 343 | |
| 344 | # Handle security |
| 345 | if "security" in endpoint: |
| 346 | security = endpoint.get("security") |
| 347 | else: |
| 348 | security = self.spec.get("security", []) |
| 349 | |
| 350 | if self.auth and security: |
| 351 | try: |
| 352 | security_definitions = self.spec.get("securityDefinitions", {}) |
| 353 | for statement in security: |
| 354 | declaration, options = statement.copy().popitem() |
| 355 | definition = security_definitions[declaration] |
| 356 | |
| 357 | if definition["type"] == "apiKey": |
| 358 | if definition["in"] == "header": |
| 359 | token = req.headers.get(definition["name"]) |
| 360 | elif definition["in"] == "query": |
| 361 | token = req.GET.get(definition["name"]) |
| 362 | elif definition["in"] == "cookie": |
| 363 | token = req.cookies.get(definition["name"]) |
| 364 | else: |
| 365 | token = None |
| 366 | |
| 367 | if token: |
| 368 | _, auth_func = op_resolver(definition["x-operationId"]) |
| 369 | auth_resp = auth_func(token) |
| 370 | |
| 371 | # Include information on how user authenticated inside the context |
| 372 | if "auth-token" in definition["name"].lower(): |
| 373 | auth_method = "authentication token" |
| 374 | elif "api-key" in definition["name"].lower(): |
| 375 | auth_method = "API key" |
| 376 | |
| 377 | context["user"] = User.get_by_name(auth_resp.user) |
| 378 | context["auth_info"] = { |
nothing calls this directly
no test coverage detected