Add Forgejo workflow Request integration #350

Signed-off-by: tdruez <[email protected]>

Author: tdruez

dje/admin.py

@@ -1060,6 +1060,7 @@ class DataspaceConfigurationForm(forms.ModelForm):
         "github_token",
         "gitlab_token",
         "jira_token",
+        "forgejo_token",
     ]
 
     def __init__(self, *args, **kwargs):
@@ -1125,6 +1126,14 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
                 ]
             },
         ),
+        (
+            "Forgejo Integration",
+            {
+                "fields": [
+                    "forgejo_token",
+                ]
+            },
+        ),
     ]
     can_delete = False
 

dje/migrations/0011_dataspaceconfiguration_forgejo_token_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.4 on 2025-08-07 09:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dje', '0010_dataspaceconfiguration_jira_token_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='dataspaceconfiguration',
+            name='forgejo_token',
+            field=models.CharField(blank=True, help_text='Personal access token (PAT) used to authenticate API requests for the Forgejo integration. This token must have sufficient permissions to create and update issues. Keep this token secure.', max_length=255, verbose_name='Forgejo token'),
+        ),
+        migrations.AddField(
+            model_name='dataspaceconfiguration',
+            name='sourcehut_token',
+            field=models.CharField(blank=True, help_text='Access token used to authenticate API requests for the SourceHut integration. This token must have permissions to create and update tickets. Keep this token secure.', max_length=255, verbose_name='SourceHut token'),
+        ),
+    ]

dje/models.py

@@ -559,8 +559,31 @@ class DataspaceConfiguration(models.Model):
         ),
     )
 
-    def __str__(self):
-        return f"Configuration for {self.dataspace}"
+    forgejo_token = models.CharField(
+        _("Forgejo token"),
+        max_length=255,
+        blank=True,
+        help_text=_(
+            "Personal access token (PAT) used to authenticate API requests for the "
+            "Forgejo integration. This token must have sufficient permissions to create "
+            "and update issues. Keep this token secure."
+        ),
+    )
+
+    sourcehut_token = models.CharField(
+        _("SourceHut token"),
+        max_length=255,
+        blank=True,
+        help_text=_(
+            "Access token used to authenticate API requests for the SourceHut integration. "
+            "This token must have permissions to create and update tickets. "
+            "Keep this token secure."
+        ),
+    )
+
+
+def __str__(self):
+    return f"Configuration for {self.dataspace}"
 
 
 class DataspacedQuerySet(models.QuerySet):

workflow/integrations/__init__.py

@@ -9,12 +9,14 @@
 import re
 
 from workflow.integrations.base import BaseIntegration
+from workflow.integrations.forgejo import ForgejoIntegration
 from workflow.integrations.github import GitHubIntegration
 from workflow.integrations.gitlab import GitLabIntegration
 from workflow.integrations.jira import JiraIntegration
 
 __all__ = [
     "BaseIntegration",
+    "ForgejoIntegration",
     "GitHubIntegration",
     "GitLabIntegration",
     "JiraIntegration",
@@ -23,6 +25,7 @@
     "get_class_for_platform",
 ]
 
+FORGEJO_PATTERN = re.compile(r"^https://(?:[a-zA-Z0-9.-]*forgejo[a-zA-Z0-9.-]*)/[^/]+/[^/]+/?$")
 
 GITHUB_PATTERN = re.compile(r"^https://github\.com/[^/]+/[^/]+/?$")
 
@@ -34,6 +37,7 @@
 )
 
 ISSUE_TRACKER_PATTERNS = [
+    FORGEJO_PATTERN,
     GITHUB_PATTERN,
     GITLAB_PATTERN,
     JIRA_PATTERN,
@@ -51,10 +55,13 @@ def get_class_for_tracker(issue_tracker_id):
         return GitLabIntegration
     elif "atlassian.net" in issue_tracker_id:
         return JiraIntegration
+    elif "forgejo" in issue_tracker_id:
+        return ForgejoIntegration
 
 
 def get_class_for_platform(platform):
     return {
+        "forgejo": ForgejoIntegration,
         "github": GitHubIntegration,
         "gitlab": GitLabIntegration,
         "jira": JiraIntegration,

workflow/integrations/forgejo.py

@@ -0,0 +1,99 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+from urllib.parse import urlparse
+
+from workflow.integrations.base import BaseIntegration
+
+FORGEJO_API_URL = ""
+
+
+class ForgejoIntegration(BaseIntegration):
+    """
+    A class for managing Forgejo issue creation, updates, and comments
+    from DejaCode requests.
+    """
+
+    api_url = FORGEJO_API_URL
+
+    def get_headers(self):
+        forgejo_token = self.dataspace.get_configuration(field_name="forgejo_token")
+        if not forgejo_token:
+            raise ValueError("The forgejo_token is not set on the Dataspace.")
+        return {"Authorization": f"token {forgejo_token}"}
+
+    def sync(self, request):
+        """Sync the given request with Forgejo by creating or updating an issue."""
+        try:
+            repo_id = self.extract_forgejo_repo_path(request.request_template.issue_tracker_id)
+        except ValueError as error:
+            raise ValueError(f"Invalid Forgejo repository URL: {error}")
+
+        external_issue = request.external_issue
+        if external_issue:
+            self.update_issue(
+                repo_id=repo_id,
+                issue_id=external_issue.issue_id,
+                title=self.make_issue_title(request),
+                body=self.make_issue_body(request),
+                state="closed" if request.is_closed else "open",
+            )
+        else:
+            issue = self.create_issue(
+                repo_id=repo_id,
+                title=self.make_issue_title(request),
+                body=self.make_issue_body(request),
+            )
+            request.link_external_issue(
+                platform="forgejo",
+                repo=repo_id,
+                issue_id=issue["number"],
+            )
+
+    def create_issue(self, repo_id, title, body=""):
+        """Create a new Forgejo issue."""
+        url = f"{self.api_url}/repos/{repo_id}/issues"
+        data = {
+            "title": title,
+            "body": body,
+        }
+
+        return self.post(url, json=data)
+
+    def update_issue(self, repo_id, issue_id, title=None, body=None, state=None):
+        """Update an existing Forgejo issue."""
+        url = f"{self.api_url}/repos/{repo_id}/issues/{issue_id}"
+        data = {}
+        if title:
+            data["title"] = title
+        if body:
+            data["body"] = body
+        if state:
+            data["state"] = state
+
+        return self.patch(url, json=data)
+
+    def post_comment(self, repo_id, issue_id, comment_body):
+        """Post a comment on an existing Forgejo issue."""
+        url = f"{self.api_url}/repos/{repo_id}/issues/{issue_id}/comments"
+        data = {"body": comment_body}
+
+        return self.post(url, json=data)
+
+    @staticmethod
+    def extract_forgejo_repo_path(url):
+        """Extract 'owner/repo-name' from a Forgejo URL."""
+        parsed = urlparse(url)
+        if not parsed.netloc:
+            raise ValueError("URL must include a hostname.")
+
+        path_parts = [part for part in parsed.path.split("/") if part]
+        if len(path_parts) < 2:
+            raise ValueError("Incomplete Forgejo repository path.")
+
+        return f"{path_parts[0]}/{path_parts[1]}"

workflow/migrations/0002_requesttemplate_issue_tracker_id_externalissuelink_and_more.py

@@ -23,7 +23,7 @@ class Migration(migrations.Migration):
             fields=[
                 ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                 ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, verbose_name='UUID')),
-                ('platform', models.CharField(choices=[('github', 'GitHub'), ('gitlab', 'GitLab'), ('jira', 'Jira'), ('sourcehut', 'SourceHut')], help_text='External issue tracking platform.', max_length=20)),
+                ('platform', models.CharField(choices=[('github', 'GitHub'), ('gitlab', 'GitLab'), ('jira', 'Jira'), ('sourcehut', 'SourceHut'), ('forgejo', 'Forgejo')], help_text='External issue tracking platform.', max_length=20)),
                 ('repo', models.CharField(help_text="Repository or project identifier (e.g., 'user/repo-name').", max_length=100)),
                 ('issue_id', models.CharField(help_text='ID or key of the issue on the external platform.', max_length=100)),
                 ('dataspace', models.ForeignKey(editable=False, help_text='A Dataspace is an independent, exclusive set of DejaCode data, which can be either nexB master reference data or installation-specific data.', on_delete=django.db.models.deletion.PROTECT, to='dje.dataspace')),

workflow/models.py

@@ -95,17 +95,22 @@ class Platform(models.TextChoices):
         GITLAB = "gitlab", _("GitLab")
         JIRA = "jira", _("Jira")
         SOURCEHUT = "sourcehut", _("SourceHut")
+        FORGEJO = "forgejo", _("Forgejo")
 
     platform = models.CharField(
-        max_length=20, choices=Platform.choices, help_text="External issue tracking platform."
+        max_length=20,
+        choices=Platform.choices,
+        help_text=_("External issue tracking platform."),
     )
 
     repo = models.CharField(
-        max_length=100, help_text="Repository or project identifier (e.g., 'user/repo-name')."
+        max_length=100,
+        help_text=_("Repository or project identifier (e.g., 'user/repo-name')."),
     )
 
     issue_id = models.CharField(
-        max_length=100, help_text="ID or key of the issue on the external platform."
+        max_length=100,
+        help_text=_("ID or key of the issue on the external platform."),
     )
 
     class Meta:
@@ -125,6 +130,8 @@ def issue_url(self):
             return f"https://gitlab.com/{self.repo}/-/issues/{self.issue_id}"
         elif self.platform == self.Platform.JIRA:
             return f"{self.repo}/browse/{self.issue_id}"
+        elif self.platform == self.Platform.FORGEJO:
+            return f"{self.repo}/issues/{self.issue_id}"
 
     @property
     def icon_css_class(self):

workflow/tests/test_integrations.py

@@ -15,12 +15,13 @@
 
 from dje.models import Dataspace
 from dje.tests import create_superuser
+from workflow.integrations import ForgejoIntegration
+from workflow.integrations import GitHubIntegration
+from workflow.integrations import GitLabIntegration
 from workflow.integrations import JiraIntegration
 from workflow.integrations import get_class_for_platform
 from workflow.integrations import get_class_for_tracker
 from workflow.integrations import is_valid_issue_tracker_id
-from workflow.integrations.github import GitHubIntegration
-from workflow.integrations.gitlab import GitLabIntegration
 from workflow.models import Question
 from workflow.models import RequestTemplate
 
@@ -41,6 +42,10 @@ def test_integrations_is_valid_issue_tracker_id(self):
             "https://example.atlassian.net/jira/software/projects/PROJ/",
             "https://example.atlassian.net/jira/software/projects/PROJ/summary",
             "https://example.atlassian.net/jira/servicedesk/projects/PROJ",
+            # Forgejo
+            "https://code.forgejo.org/user/repo",
+            "https://git.forgejo.dev/org/project/",
+            "https://forgejo.example.org/team/repo",
         ]
         for url in valid_urls:
             self.assertTrue(is_valid_issue_tracker_id(url), msg=url)
@@ -51,6 +56,7 @@ def test_integrations_is_valid_issue_tracker_id(self):
             "https://gitlab.com/",
             "https://atlassian.net/projects/",
             "https://example.com",
+            "https://example.org/user/repo",
         ]
         for url in invalid_urls:
             self.assertFalse(is_valid_issue_tracker_id(url), msg=url)
@@ -61,12 +67,16 @@ def test_integrations_get_class_for_tracker(self):
         self.assertIs(
             get_class_for_tracker("https://example.atlassian.net/projects/PROJ"), JiraIntegration
         )
+        self.assertIs(
+            get_class_for_tracker("https://code.forgejo.org/user/repo"), ForgejoIntegration
+        )
         self.assertIsNone(get_class_for_tracker("https://example.com"))
 
     def test_integrations_get_class_for_platform(self):
         self.assertIs(get_class_for_platform("github"), GitHubIntegration)
         self.assertIs(get_class_for_platform("gitlab"), GitLabIntegration)
         self.assertIs(get_class_for_platform("jira"), JiraIntegration)
+        self.assertIs(get_class_for_platform("forgejo"), ForgejoIntegration)
         self.assertIsNone(get_class_for_platform("example"))