pufereq/kursy-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(signals.py): create per-lesson repositories on ModuleLessonPage save

Browse Source
This commit is contained in:
Artur Borecki
2026-03-30 10:58:01 +02:00
parent 787440d56f
commit dd936473d8
1 changed files with 179 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

283
home/signals.py
View File

@@ -6,7 +6,7 @@ from django.conf import settings
from django.db.models.signals import post_save from django.db.models.signals import post_save
from django.dispatch import receiver from django.dispatch import receiver
from home.models.pages import CoursePage from home.models.pages import CoursePage, ModuleLessonPage
GITEA_ORG_NAME = "Studio77" GITEA_ORG_NAME = "Studio77"
@@ -14,7 +14,7 @@ logger = lg.getLogger(__name__)
@receiver(post_save, sender=CoursePage) @receiver(post_save, sender=CoursePage)
def create_gitea_team_repo_on_course_creation(sender, instance, created, **kwargs): def create_gitea_team_on_course_creation(sender, instance, created, **kwargs):
if not instance.live: if not instance.live:
logger.debug( logger.debug(
f"Course {instance.title} is not live, skipping Gitea team creation" f"Course {instance.title} is not live, skipping Gitea team creation"
@@ -29,113 +29,188 @@ def create_gitea_team_repo_on_course_creation(sender, instance, created, **kwarg
logger.debug("GITEA_URL is not set, skipping Gitea team creation") logger.debug("GITEA_URL is not set, skipping Gitea team creation")
return return
def team(): # check if team already exists
# check if team already exists try:
try: response = requests.get(
response = requests.get( f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
f"{api_url}/orgs/{GITEA_ORG_NAME}/teams", timeout=5,
timeout=5, headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"}, )
) response.raise_for_status()
response.raise_for_status() teams = response.json()
teams = response.json() if any(team["name"] == team_name for team in teams):
if any(team["name"] == team_name for team in teams): logger.info(f"Gitea team {team_name} already exists, skipping creation")
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 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" url = f"{api_url}/orgs/{GITEA_ORG_NAME}/teams"
payload = { payload = {
"can_create_org_repo": False, "can_create_org_repo": False,
"description": f"Team for course {course.title}", "description": f"Team for course {course.title}",
"includes_all_repositories": False, "includes_all_repositories": False,
"name": team_name, "name": team_name,
"permission": "read", "permission": "read",
"units": [ "units": [
# "repo.actions", # "repo.actions",
"repo.code", "repo.code",
# "repo.issues", # "repo.issues",
# "repo.ext_issues", # "repo.ext_issues",
# "repo.wiki", # "repo.wiki",
# "repo.ext_wiki", # "repo.ext_wiki",
# "repo.pulls", # "repo.pulls",
# "repo.releases", # "repo.releases",
# "repo.projects", # "repo.projects",
# "repo.ext_wiki", # "repo.ext_wiki",
], ],
} }
try: try:
response = requests.post( response = requests.post(
url, url,
json=payload, json=payload,
timeout=5, timeout=5,
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"}, headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
) )
response.raise_for_status() response.raise_for_status()
logger.info(f"Successfully created Gitea team for course {course.title}") logger.info(f"Successfully created Gitea team for course {course.title}")
except Exception as e: except Exception as e:
logger.exception( logger.exception(
f"Failed to create Gitea team for course {course.title}: {e}\n{response.text}", f"Failed to create Gitea team for course {course.title}: {e}\n{response.text}",
e, e,
) )
def repo(): # def repo():
# check if repository already exists # # check if repository already exists
try: # try:
response = requests.get( # response = requests.get(
f"{api_url}/orgs/{GITEA_ORG_NAME}/repos", # f"{api_url}/orgs/{GITEA_ORG_NAME}/repos",
timeout=5, # timeout=5,
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"}, # headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
) # )
response.raise_for_status() # response.raise_for_status()
repos = response.json() # repos = response.json()
if any(repo["name"] == team_name for repo in repos): # if any(repo["name"] == team_name for repo in repos):
logger.debug( # logger.debug(
f"Gitea repository {team_name} already exists, skipping creation" # f"Gitea repository {team_name} already exists, skipping creation"
) # )
return # return
except Exception as e: # except Exception as e:
logger.exception( # logger.exception(
f"Failed to check existing Gitea repositories: {e}\n{response.text}", # f"Failed to check existing Gitea repositories: {e}\n{response.text}",
e, # e,
) # )
return # 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,
# )
#
#
# 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: @receiver(post_save, sender=ModuleLessonPage)
response = requests.post( def create_gitea_repo_on_lesson_creation(
url, sender, instance: ModuleLessonPage, created, **kwargs
json=payload, ):
timeout=5, if not instance.live:
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"}, logger.debug(
) f"Lesson {instance.title} is not live, skipping Gitea repository creation"
response.raise_for_status() )
repo_url = response.json().get("url", None) return
course.repository_url = repo_url
course.save(update_fields=["repository_url"]) course = instance.module.course
repo_name = f"course-{course.id}-lesson-{instance.id}"
if not course.live:
logger.debug(
f"Course {course.title} is not live, skipping Gitea repository creation for lesson {instance.title}"
)
return
api_url = getattr(settings, "GITEA_URL", None)
if not api_url:
logger.debug("GITEA_URL is not set, skipping Gitea repository creation")
return
# 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"] == repo_name for repo in repos):
logger.info( logger.info(
f"Successfully created Gitea repository for course {course.title}" f"Gitea repository {repo_name} already exists, skipping creation"
)
except Exception as e:
logger.exception(
f"Failed to create Gitea repository for course {course.title}: {e}\n{response.text}",
e,
) )
return
except Exception as e:
logger.exception(
f"Failed to check existing Gitea repositories: {e}\n{response.text}",
e,
)
return
team() # create lesson repository
repo() url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
payload = {
"auto_init": True,
"default_branch": "main",
"description": f"{instance.module.course} – {instance.module}: {instance.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)
instance.gitea_repo_url = repo_url
instance.save()
logger.info(
f"Successfully created Gitea repository for lesson {instance.title}"
)
except Exception as e:
logger.exception(
f"Failed to create Gitea repository for lesson {instance.title}: {e}\n{response.text}",
e,
)