From 6dd826c3bd72ffcdc287ae30b1511a8084e23ebe Mon Sep 17 00:00:00 2001 From: Artur Borecki Date: Mon, 23 Mar 2026 14:02:07 +0100 Subject: [PATCH] feat(home/signals.py): create gitea team and repo for course on CoursePage save --- home/signals.py | 164 +++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 133 insertions(+), 31 deletions(-) diff --git a/home/signals.py b/home/signals.py index 73b9847..d44dc75 100644 --- a/home/signals.py +++ b/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) -def notify_external_service_on_signup(sender, instance, created, **kwargs): - pass - # if created and not instance.is_staff: - # payload = { - # "user_id": instance.id, - # "username": f"KURSY-{instance.id}", - # "email": instance.email, - # "full_name": f"{instance.first_name} {instance.last_name}".strip(), - # # "must_change_password": True, - # # "password": instance.password, - # "visibility": "private", - # } - # api_url = getattr(settings, "GITEA_URL", None) - # if api_url: - # url = f"{api_url}/admin/users" - # try: - # response = requests.post( - # url, - # json=payload, - # timeout=5, - # headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"}, - # ) - # response.raise_for_status() - # print(f"Successfully created Gitea account for {instance.email}") - # except Exception as e: - # print( - # f"Failed to create Gitea account for user {instance.email}: {e}\n{response.text}" - # ) - # raise e +GITEA_ORG_NAME = "Studio77" + +logger = lg.getLogger(__name__) + + +@receiver(post_save, sender=CoursePage) +def create_gitea_team_repo_on_course_creation(sender, instance, created, **kwargs): + if not instance.live: + logger.debug( + f"Course {instance.title} is not live, skipping Gitea team creation" + ) + return + + course = instance + 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 creation") + return + + def team(): + # check if team already exists + 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() + 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()