feat(home/signals.py): create gitea team and repo for course on CoursePage save

home/signals.py

@@ -1,39 +1,141 @@
+import logging as lg
 import os
 
 import requests
 from django.conf import settings
-from django.contrib.auth.models import User
 from django.db.models.signals import post_save
 from django.dispatch import receiver
 
+from home.models.pages import CoursePage
 
-@receiver(post_save, sender=User)
+GITEA_ORG_NAME = "Studio77"
-def notify_external_service_on_signup(sender, instance, created, **kwargs):
+
-    pass
+logger = lg.getLogger(__name__)
-    # if created and not instance.is_staff:
+
-    #     payload = {
+
-    #         "user_id": instance.id,
+@receiver(post_save, sender=CoursePage)
-    #         "username": f"KURSY-{instance.id}",
+def create_gitea_team_repo_on_course_creation(sender, instance, created, **kwargs):
-    #         "email": instance.email,
+    if not instance.live:
-    #         "full_name": f"{instance.first_name} {instance.last_name}".strip(),
+        logger.debug(
-    #         # "must_change_password": True,
+            f"Course {instance.title} is not live, skipping Gitea team creation"
-    #         # "password": instance.password,
+        )
-    #         "visibility": "private",
+        return
-    #     }
+
-    #     api_url = getattr(settings, "GITEA_URL", None)
+    course = instance
-    #     if api_url:
+    team_name = f"course-{course.id}"
-    #         url = f"{api_url}/admin/users"
+    api_url = getattr(settings, "GITEA_URL", None)
-    #         try:
+
-    #             response = requests.post(
+    if not api_url:
-    #                 url,
+        logger.debug("GITEA_URL is not set, skipping Gitea team creation")
-    #                 json=payload,
+        return
-    #                 timeout=5,
+
-    #                 headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+    def team():
-    #             )
+        # check if team already exists
-    #             response.raise_for_status()
+        try:
-    #             print(f"Successfully created Gitea account for {instance.email}")
+            response = requests.get(
-    #         except Exception as e:
+                f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
-    #             print(
+                timeout=5,
-    #                 f"Failed to create Gitea account for user {instance.email}: {e}\n{response.text}"
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
-    #             )
+            )
-    #             raise e
+            response.raise_for_status()
+            teams = response.json()
+            if any(team["name"] == team_name for team in teams):
+                logger.info(f"Gitea team {team_name} already exists, skipping creation")
+                return
+        except Exception as e:
+            logger.exception(
+                f"Failed to check existing Gitea teams: {e}\n{response.text}",
+                e,
+            )
+            return
+
+        url = f"{api_url}/orgs/{GITEA_ORG_NAME}/teams"
+        payload = {
+            "can_create_org_repo": False,
+            "description": f"Team for course {course.title}",
+            "includes_all_repositories": False,
+            "name": team_name,
+            "permission": "read",
+            "units": [
+                # "repo.actions",
+                "repo.code",
+                # "repo.issues",
+                # "repo.ext_issues",
+                # "repo.wiki",
+                # "repo.ext_wiki",
+                # "repo.pulls",
+                # "repo.releases",
+                # "repo.projects",
+                # "repo.ext_wiki",
+            ],
+        }
+        try:
+            response = requests.post(
+                url,
+                json=payload,
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            response.raise_for_status()
+            logger.info(f"Successfully created Gitea team for course {course.title}")
+        except Exception as e:
+            logger.exception(
+                f"Failed to create Gitea team for course {course.title}: {e}\n{response.text}",
+                e,
+            )
+
+    def repo():
+        # check if repository already exists
+        try:
+            response = requests.get(
+                f"{api_url}/orgs/{GITEA_ORG_NAME}/repos",
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            response.raise_for_status()
+            repos = response.json()
+            if any(repo["name"] == team_name for repo in repos):
+                logger.debug(
+                    f"Gitea repository {team_name} already exists, skipping creation"
+                )
+                return
+        except Exception as e:
+            logger.exception(
+                f"Failed to check existing Gitea repositories: {e}\n{response.text}",
+                e,
+            )
+            return
+
+        # create course repository
+        repo_name = f"course-{course.id}"
+        url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
+        payload = {
+            "auto_init": True,
+            "default_branch": "main",
+            "description": f"{course.title}",
+            "name": repo_name,
+            "private": True,
+        }
+
+        try:
+            response = requests.post(
+                url,
+                json=payload,
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            response.raise_for_status()
+            repo_url = response.json().get("url", None)
+            course.repository_url = repo_url
+            course.save(update_fields=["repository_url"])
+            logger.info(
+                f"Successfully created Gitea repository for course {course.title}"
+            )
+        except Exception as e:
+            logger.exception(
+                f"Failed to create Gitea repository for course {course.title}: {e}\n{response.text}",
+                e,
+            )
+
+    team()
+    repo()