(self, request: Request, call_next: RequestResponseEndpoint)
| 20 | self.hash_service = hash_service |
| 21 | |
| 22 | async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: |
| 23 | if not self._is_included_path(request.url.path): |
| 24 | return await call_next(request) |
| 25 | |
| 26 | authorization = request.headers.get("X-Api-Key") |
| 27 | if not authorization: |
| 28 | raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_X_API_KEY_EMPTY, message="Unauthorized!") |
| 29 | |
| 30 | if not self.hash_service.verify_x_api_key(key=authorization): |
| 31 | raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_X_API_KEY_INVALID, message="Unauthorized!") |
| 32 | |
| 33 | return await call_next(request) |
| 34 | |
| 35 | @staticmethod |
| 36 | def _is_included_path(path: str) -> bool: |
nothing calls this directly
no test coverage detected