Files
kursy-mirror/home/signals.py

142 lines
4.5 KiB
Python

import logging as lg
import os
import requests
from django.conf import settings
from django.db.models.signals import post_save
from django.dispatch import receiver
from home.models.pages import CoursePage
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()