Changeset View
Changeset View
Standalone View
Standalone View
swh/web/save_origin_webhooks/generic_receiver.py
| Show All 19 Lines | class OriginSaveWebhookReceiver(abc.ABC): | ||||
| FORGE_TYPE: str | FORGE_TYPE: str | ||||
| WEBHOOK_GUIDE_URL: str | WEBHOOK_GUIDE_URL: str | ||||
| REPO_TYPES: str | REPO_TYPES: str | ||||
| @abc.abstractmethod | @abc.abstractmethod | ||||
| def is_forge_request(self, request: Request) -> bool: | def is_forge_request(self, request: Request) -> bool: | ||||
| ... | ... | ||||
| def is_ping_event(self, request: Request) -> bool: | |||||
| return False | |||||
| @abc.abstractmethod | @abc.abstractmethod | ||||
| def is_push_event(self, request: Request) -> bool: | def is_push_event(self, request: Request) -> bool: | ||||
| ... | ... | ||||
| @abc.abstractmethod | @abc.abstractmethod | ||||
| def extract_repo_url_and_visit_type(self, request: Request) -> Tuple[str, str]: | def extract_repo_url_and_visit_type(self, request: Request) -> Tuple[str, str]: | ||||
| ... | ... | ||||
| Show All 40 Lines | class OriginSaveWebhookReceiver(abc.ABC): | ||||
| ) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
| if not self.is_forge_request(request): | if not self.is_forge_request(request): | ||||
| raise BadInputExc( | raise BadInputExc( | ||||
| f"POST request was not sent by a {self.FORGE_TYPE} webhook and " | f"POST request was not sent by a {self.FORGE_TYPE} webhook and " | ||||
| "has not been processed." | "has not been processed." | ||||
| ) | ) | ||||
| if self.is_ping_event(request): | |||||
| return {"message": "pong"} | |||||
| if not self.is_push_event(request): | if not self.is_push_event(request): | ||||
| raise BadInputExc( | raise BadInputExc( | ||||
| f"Event sent by {self.FORGE_TYPE} webhook is not a push one, request " | f"Event sent by {self.FORGE_TYPE} webhook is not a push one, request " | ||||
| "has not been processed." | "has not been processed." | ||||
| ) | ) | ||||
| content_type = request.headers.get("Content-Type") | content_type = request.headers.get("Content-Type") | ||||
| if content_type and not content_type.startswith("application/json"): | if content_type and not content_type.startswith("application/json"): | ||||
| Show All 26 Lines | |||||