From 12fdc1881f93ec1f2207d4593dbb3ac14928b081 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Brunner?= Date: Mon, 18 Nov 2024 11:12:10 +0100 Subject: [PATCH] Add module used to dispatch publishing events --- github_app_geo_project/module/__init__.py | 7 + .../module/dispatch_publishing/__init__.py | 137 ++++++++++++++++++ github_app_geo_project/views/project.py | 2 +- pyproject.toml | 1 + 4 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 github_app_geo_project/module/dispatch_publishing/__init__.py diff --git a/github_app_geo_project/module/__init__.py b/github_app_geo_project/module/__init__.py index 47f6e0ecf4..4161a54403 100644 --- a/github_app_geo_project/module/__init__.py +++ b/github_app_geo_project/module/__init__.py @@ -364,6 +364,13 @@ def cleanup(self, context: CleanupContext[_EVENT_DATA]) -> None: @abstractmethod def get_json_schema(self) -> dict[str, Any]: """Get the JSON schema of the module configuration.""" + super_ = [c for c in self.__class__.__orig_bases__ if c.__origin__ == Module][0] # type: ignore[attr-defined] # pylint: disable=no-member + generic_element = super_.__args__[0] + # Is Pydantic BaseModel + if not isinstance(generic_element, GenericAlias) and issubclass(generic_element, BaseModel): + return generic_element.model_json_schema() # type: ignore[no-any-return] + else: + raise NotImplementedError("The method get_json_schema should be implemented") def configuration_from_json(self, data: dict[str, Any]) -> _CONFIGURATION: """Create the configuration from the JSON data.""" diff --git a/github_app_geo_project/module/dispatch_publishing/__init__.py b/github_app_geo_project/module/dispatch_publishing/__init__.py new file mode 100644 index 0000000000..d559010dd4 --- /dev/null +++ b/github_app_geo_project/module/dispatch_publishing/__init__.py @@ -0,0 +1,137 @@ +"""Module to dispatch publishing event.""" + +import json +import logging +import os +import re +from typing import Any + +from pydantic import BaseModel + +from github_app_geo_project import module + +_LOGGER = logging.getLogger(__name__) + + +class _Destination(BaseModel): + """The destination to dispatch to.""" + + destination_repository: str + """The repository to dispatch to""" + event_type: str + """The event type to dispatch""" + legacy: bool = False + """Transform the content to the legacy format""" + version_type: str + """The version type to dispatch""" + package_type: str + """The package type to dispatch""" + image_re: str = ".*" + """The image regular expression to dispatch""" + + +class _Config(BaseModel): + """The configuration of the module.""" + + destinations: list[_Destination] = [] + """The destinations to dispatch to""" + + +CONFIG = _Config(**json.loads(os.environ.get("DISPATCH_PUBLISH_CONFIG", "{}"))) + + +class DispatchPublishing(module.Module[None, None, None]): + """ + The version module. + + Create a dashboard to show the back ref versions with support check + """ + + def title(self) -> str: + """Get the title of the module.""" + return "Dispatch" + + def description(self) -> str: + """Get the description of the module.""" + return "Dispatch publishing event" + + def documentation_url(self) -> str: + """Get the URL to the documentation page of the module.""" + return "https://github.com/camptocamp/github-app-geo-project/wiki/Module-%E2%80%90-Dispatch-Publish" + + def get_json_schema(self) -> dict[str, Any]: + """Get the JSON schema for the configuration.""" + return {} + + def get_actions(self, context: module.GetActionContext) -> list[module.Action[None]]: + """ + Get the action related to the module and the event. + + Usually the only action allowed to be done in this method is to set the pull request checks status + Note that this function is called in the web server Pod who has low resources, and this call should be fast + """ + if context.event_name == "repository_dispatch" and context.event_data.get("event_type") == "publish": + return [module.Action(None)] + return [] + + def get_github_application_permissions(self) -> module.GitHubApplicationPermissions: + """Get the GitHub application permissions needed by the module.""" + return module.GitHubApplicationPermissions( + permissions={"contents": "write"}, events={"repository_dispatch"} + ) + + async def process( + self, + context: module.ProcessContext[None, None, None], + ) -> module.ProcessOutput[None, None]: + """ + Process the action. + + Note that this method is called in the queue consuming Pod + """ + for destination in CONFIG.destinations: + content = context.event_data.get("payloads", {}).get("content", {}) + if destination.version_type and destination.package_type != content.get("version_type"): + continue + + image_re = re.compile(destination.image_re) + payload: dict[str, Any] = {} + names = [] + + for item in content.get("items", []): + if destination.package_type and destination.package_type != item.package_type: + continue + + if not image_re.match(item.get("image", "")): + continue + + if destination.legacy: + if "image" in item: + if item.get("repository", "") in ("", "docker.io"): + names.append(item["image"]) + else: + names.append(f'{item["repository"]}/{item["image"]}') + else: + payload.setdefault("content", {}).setdefault("items", []).append(item) + + if destination.legacy and names: + payload["name"] = " ".join(names) + + if payload: + context.github_project.github.get_repo( + destination.destination_repository + ).create_repository_dispatch( + destination.event_type, + payload, + ) + return module.ProcessOutput() + + destination_repository: str + legacy: bool = False + version_type: str + package_type: str + image_re: str = ".*" + + def has_transversal_dashboard(self) -> bool: + """Return True if the module has a transversal dashboard.""" + return False diff --git a/github_app_geo_project/views/project.py b/github_app_geo_project/views/project.py index 78a06d84d7..9c6fe54de5 100644 --- a/github_app_geo_project/views/project.py +++ b/github_app_geo_project/views/project.py @@ -96,7 +96,7 @@ def project(request: pyramid.request.Request) -> dict[str, Any]: if module.required_issue_dashboard(): applications[app]["issue_required"] = True - except: # nosec, pylint: disable=bare-except + except: # pylint: disable=bare-except _LOGGER.debug( "The repository %s/%s is not installed in the application %s", owner, repository, app ) diff --git a/pyproject.toml b/pyproject.toml index 4df0d51e8b..5f04ff18cf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ workflow = "github_app_geo_project.module.workflow:Workflow" pull-request-checks = "github_app_geo_project.module.pull_request.checks:Checks" pull-request-links = "github_app_geo_project.module.pull_request.links:Links" delete-old-workflow-runs = "github_app_geo_project.module.delete_old_workflow_runs:DeleteOldWorkflowRuns" +dispatch-publishing = "github_app_geo_project.module.dispatch_publishing:DispatchPublishing" [tool.poetry.dependencies] python = ">=3.11,<3.13"