feat(models.py): add users to gitea teams on purchase/refund

purchase/models.py

@@ -1,7 +1,14 @@
+import logging as lg
+import os
+
+import requests
 from django.conf import settings
 from django.contrib.auth.models import Group
 from django.db import models
 
+GITEA_ORG_NAME = "Studio77"
+logger = lg.getLogger(__name__)
+
 
 class CoursePurchase(models.Model):
     user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
@@ -13,11 +20,115 @@ class CoursePurchase(models.Model):
         self.refunded = True
         self.save()
 
+    def _get_gitea_team_id(self, team_name):
+        api_url = getattr(settings, "GITEA_URL", None)
+        if not api_url:
+            logger.debug("GITEA_URL is not set, skipping Gitea team assignment")
+            return None
+
+        try:
+            response = requests.get(
+                f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            response.raise_for_status()
+            teams = response.json()
+            team = next((team for team in teams if team["name"] == team_name), None)
+            return team["id"] if team else None
+        except Exception as e:
+            logger.exception(
+                f"Failed to check existing Gitea teams: {e}\n{getattr(response, 'text', '')}",
+                e,
+            )
+            return None
+
+    def add_to_gitea_team(self):
+        course = self.course
+        user = self.user
+        team_name = f"course-{course.id}"
+        api_url = getattr(settings, "GITEA_URL", None)
+
+        if not api_url:
+            logger.debug("GITEA_URL is not set, skipping Gitea team assignment")
+            return
+
+        team_id = self._get_gitea_team_id(team_name)
+        if not team_id:
+            logger.warning(
+                f"Gitea team {team_name} not found for course {course.title}"
+            )
+            return
+
+        url = f"{api_url}/teams/{team_id}/members/studio77-{user.id}"
+        try:
+            response = requests.put(
+                url,
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            if response.status_code == 204:
+                logger.info(
+                    f"Successfully added user {user.email} to Gitea team {team_name}"
+                )
+            else:
+                response.raise_for_status()
+        except Exception as e:
+            logger.error(
+                f"Failed to add user {user.email} to Gitea team {team_name}: {e}\n{getattr(response, 'text', '')}"
+            )
+
+    def remove_from_gitea_team(self):
+        course = self.course
+        user = self.user
+        team_name = f"course-{course.id}"
+        api_url = getattr(settings, "GITEA_URL", None)
+
+        if not api_url:
+            logger.debug("GITEA_URL is not set, skipping Gitea team removal")
+            return
+
+        team_id = self._get_gitea_team_id(team_name)
+        if not team_id:
+            logger.warning(
+                f"Gitea team {team_name} not found for course {course.title}"
+            )
+            return
+
+        url = f"{api_url}/teams/{team_id}/members/studio77-{user.id}"
+        try:
+            response = requests.delete(
+                url,
+                timeout=5,
+                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
+            )
+            if response.status_code == 204:
+                logger.info(
+                    f"Successfully removed user {user.email} from Gitea team {team_name}"
+                )
+            else:
+                response.raise_for_status()
+        except Exception as e:
+            logger.error(
+                f"Failed to remove user {user.email} from Gitea team {team_name}: {e}\n{getattr(response, 'text', '')}"
+            )
+
     def save(self, *args, **kwargs):
         super().save(*args, **kwargs)
         group_name = f"course_{self.course.id}_access"
 
         group, _ = Group.objects.get_or_create(name=group_name)
+
+        print(
+            f"Saving CoursePurchase for user {self.user} and course {self.course.title}, refunded={self.refunded}"
+        )
+
         if self.refunded:
             print(f"Removing user {self.user} from group {group_name} due to refund")
+            self.remove_from_gitea_team()
             self.user.groups.remove(group)
+        else:
+            logger.debug(
+                f"Adding user {self.user} to group {group_name} for course {self.course.title}"
+            )
+            self.add_to_gitea_team()