(self, method: str, *args, **kwargs)
| 89 | return super().send(request, **kwargs) |
| 90 | |
| 91 | def request(self, method: str, *args, **kwargs): |
| 92 | callee = type(self).__name__ |
| 93 | if logger.getEffectiveLevel() >= logging.DEBUG: |
| 94 | logger.debug("[%s] request(%r, *%r, ...)", |
| 95 | callee, method, args) |
| 96 | else: # pragma: no cover |
| 97 | logger.trace("[%s] request(%r, *%r, **%r)", |
| 98 | callee, method, args, kwargs) |
| 99 | |
| 100 | try: |
| 101 | response = self._request_unified(method, *args, **kwargs) |
| 102 | |
| 103 | rec = LoggingSession.RequestRecord( |
| 104 | request=response.request, response=response, exception=None) |
| 105 | self.history.append(rec) |
| 106 | |
| 107 | except Exception as exc: |
| 108 | logger.debug("[%s] request failed with %r", callee, exc) |
| 109 | |
| 110 | req = getattr(exc, 'request', None) |
| 111 | if req is not None: |
| 112 | logger.trace("[%s] failing request=%r", callee, |
| 113 | dict(method=req.method, url=req.url, |
| 114 | headers=req.headers, body=req.body)) |
| 115 | |
| 116 | res = getattr(exc, 'response', None) |
| 117 | if res is not None: |
| 118 | logger.trace("[%s] failing response=%r", callee, |
| 119 | dict(status_code=res.status_code, |
| 120 | headers=res.headers, text=res.text)) |
| 121 | |
| 122 | rec = LoggingSession.RequestRecord( |
| 123 | request=req, response=res, exception=exc) |
| 124 | self.history.append(rec) |
| 125 | |
| 126 | raise |
| 127 | |
| 128 | if logger.getEffectiveLevel() >= logging.DEBUG: |
| 129 | logger.debug("[%s] request succeeded with status_code=%r", |
| 130 | callee, response.status_code) |
| 131 | else: # pragma: no cover |
| 132 | logger.trace("[%s] request(...) = (code=%r, headers=%r, body=%r)", |
| 133 | callee, response.status_code, response.headers, response.text) |
| 134 | |
| 135 | return response |
| 136 | |
| 137 | |
| 138 | class LoggingSession(LoggingSessionMixin, BaseUrlSessionMixin, requests.Session): |
no test coverage detected