659 lines
22 KiB
Python
659 lines
22 KiB
Python
import base64
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import date
|
|
from enum import Enum
|
|
from typing import List, Set, Tuple
|
|
|
|
from django.db.models import Q
|
|
from django.http import HttpResponse
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.response import Response
|
|
|
|
from vbv_lernwelt.assignment.export import (
|
|
COMPETENCE_ELEMENT_EXPORT_FILE_NAME,
|
|
export_competence_elements,
|
|
)
|
|
from vbv_lernwelt.assignment.models import (
|
|
AssignmentCompletion,
|
|
AssignmentCompletionStatus,
|
|
)
|
|
from vbv_lernwelt.competence.services import (
|
|
query_competence_course_session_assignments,
|
|
query_competence_course_session_edoniq_tests,
|
|
)
|
|
from vbv_lernwelt.core.models import User
|
|
from vbv_lernwelt.course.models import (
|
|
CourseConfiguration,
|
|
CourseSession,
|
|
CourseSessionUser,
|
|
)
|
|
from vbv_lernwelt.course.views import logger
|
|
from vbv_lernwelt.course_session.services.export_attendance import (
|
|
ATTENDANCE_EXPORT_FILENAME,
|
|
export_attendance,
|
|
make_export_filename,
|
|
)
|
|
from vbv_lernwelt.course_session_group.models import CourseSessionGroup
|
|
from vbv_lernwelt.duedate.models import DueDate
|
|
from vbv_lernwelt.duedate.serializers import DueDateSerializer
|
|
from vbv_lernwelt.feedback.export import (
|
|
export_feedback_with_circle_restriction,
|
|
FEEDBACK_EXPORT_FILE_NAME,
|
|
)
|
|
from vbv_lernwelt.learning_mentor.models import LearningMentor
|
|
from vbv_lernwelt.learnpath.models import Circle
|
|
from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback
|
|
|
|
|
|
class WidgetType(Enum):
|
|
PROGRESS_WIDGET = "ProgressWidget"
|
|
COMPETENCE_WIDGET = "CompetenceWidget"
|
|
MENTOR_TASKS_WIDGET = "MentorTasksWidget"
|
|
MENTOR_PERSON_WIDGET = "MentorPersonWidget"
|
|
MENTOR_COMPETENCE_WIDGET = "MentorCompetenceWidget"
|
|
COMPETENCE_CERTIFICATE_WIDGET = "CompetenceCertificateWidget"
|
|
UK_STATISTICS_WIDGET = "UKStatisticsWidget"
|
|
|
|
|
|
class RoleKeyType(Enum):
|
|
MEMBER = "Member"
|
|
MENTOR_VV = "MentorVV"
|
|
MENTOR_UK = "MentorUK"
|
|
SUPERVISOR = "Supervisor"
|
|
TRAINER = "Trainer"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CourseSessionWithRoles:
|
|
_original: CourseSession
|
|
roles: Set[str]
|
|
|
|
def __getattr__(self, name: str):
|
|
# Delegate attribute access to the _original CourseSession object
|
|
return getattr(self._original, name)
|
|
|
|
def save(self, *args, **kwargs):
|
|
raise NotImplementedError("This proxy object cannot be saved.")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CourseConfig:
|
|
course_id: str
|
|
course_slug: str
|
|
course_title: str
|
|
role_key: str
|
|
is_uk: bool
|
|
is_vv: bool
|
|
is_mentor: bool
|
|
widgets: List[str]
|
|
has_preview: bool
|
|
session_to_continue_id: str | None
|
|
|
|
|
|
def get_course_sessions_with_roles_for_user(user: User) -> List[CourseSessionWithRoles]:
|
|
result_course_sessions = {}
|
|
|
|
# participant/member/expert course sessions
|
|
csu_qs = CourseSessionUser.objects.filter(user=user).prefetch_related(
|
|
"course_session", "course_session__course"
|
|
)
|
|
for csu in csu_qs:
|
|
cs = csu.course_session
|
|
# member/expert is mutually exclusive...
|
|
cs.roles = {csu.role}
|
|
result_course_sessions[cs.id] = cs
|
|
|
|
# enrich with supervisor course sessions
|
|
csg_qs = CourseSessionGroup.objects.filter(supervisor=user).prefetch_related(
|
|
"course_session", "course_session__course"
|
|
)
|
|
for csg in csg_qs:
|
|
for cs in csg.course_session.all():
|
|
cs.roles = set()
|
|
cs = result_course_sessions.get(cs.id, cs)
|
|
|
|
cs.roles.add("SUPERVISOR")
|
|
result_course_sessions[cs.id] = cs
|
|
|
|
# enrich with mentor course sessions
|
|
lm_qs = LearningMentor.objects.filter(mentor=user).prefetch_related(
|
|
"course_session", "course_session__course"
|
|
)
|
|
for lm in lm_qs:
|
|
cs = lm.course_session
|
|
cs.roles = set()
|
|
cs = result_course_sessions.get(cs.id, cs)
|
|
|
|
cs.roles.add("LEARNING_MENTOR")
|
|
result_course_sessions[cs.id] = cs
|
|
|
|
return [
|
|
CourseSessionWithRoles(cs, cs.roles) for cs in result_course_sessions.values()
|
|
]
|
|
|
|
|
|
def has_cs_role(roles: Set[str]) -> bool:
|
|
return bool(roles & {"SUPERVISOR", "EXPERT", "MEMBER"})
|
|
|
|
|
|
def user_role(roles: Set[str]) -> str:
|
|
if "SUPERVISOR" in roles:
|
|
return "SUPERVISOR"
|
|
if "EXPERT" in roles:
|
|
return "EXPERT"
|
|
if "MEMBER" in roles:
|
|
return "MEMBER"
|
|
return "LEARNING_MENTOR"
|
|
|
|
|
|
def _create_course_session_dict(course_session_object, my_role, user_role):
|
|
return {
|
|
"id": str(course_session_object.id),
|
|
"session_title": course_session_object.title,
|
|
"course_id": str(course_session_object.course.id),
|
|
"course_title": course_session_object.course.title,
|
|
"course_slug": course_session_object.course.slug,
|
|
"region": course_session_object.region,
|
|
"generation": course_session_object.generation,
|
|
"my_role": my_role,
|
|
"user_role": user_role,
|
|
"is_uk": course_session_object.course.configuration.is_uk,
|
|
"is_vv": course_session_object.course.configuration.is_vv,
|
|
}
|
|
|
|
|
|
def _create_person_list_with_roles(user):
|
|
def create_user_dict(user_object):
|
|
return {
|
|
"user_id": user_object.id,
|
|
"first_name": user_object.first_name,
|
|
"last_name": user_object.last_name,
|
|
"email": user_object.email,
|
|
"avatar_url_small": user_object.avatar_url_small,
|
|
"avatar_url": user_object.avatar_url,
|
|
"course_sessions": [],
|
|
}
|
|
|
|
course_sessions = get_course_sessions_with_roles_for_user(user)
|
|
|
|
result_persons = {}
|
|
for cs in course_sessions:
|
|
if has_cs_role(cs.roles) and cs.course.configuration.is_uk:
|
|
course_session_users = CourseSessionUser.objects.filter(
|
|
course_session=cs.id
|
|
)
|
|
my_role = user_role(cs.roles)
|
|
for csu in course_session_users:
|
|
person_data = result_persons.get(
|
|
csu.user.id, create_user_dict(csu.user)
|
|
)
|
|
person_data["course_sessions"].append(
|
|
_create_course_session_dict(cs, my_role, csu.role)
|
|
)
|
|
result_persons[csu.user.id] = person_data
|
|
|
|
# add persons where request.user is mentor
|
|
for cs in course_sessions:
|
|
if "LEARNING_MENTOR" in cs.roles:
|
|
lm = LearningMentor.objects.filter(
|
|
mentor=user, course_session=cs.id
|
|
).first()
|
|
|
|
for participant in lm.participants.all():
|
|
course_session_entry = _create_course_session_dict(
|
|
cs,
|
|
"LEARNING_MENTOR",
|
|
"LEARNING_MENTEE",
|
|
)
|
|
|
|
if participant.user.id not in result_persons:
|
|
person_data = create_user_dict(participant.user)
|
|
person_data["course_sessions"] = [course_session_entry]
|
|
result_persons[participant.user.id] = person_data
|
|
else:
|
|
# user is already in result_persons
|
|
result_persons[participant.user.id]["course_sessions"].append(
|
|
course_session_entry
|
|
)
|
|
|
|
# add persons where request.user is mentee
|
|
mentor_relation_qs = LearningMentor.objects.filter(
|
|
participants__user=user
|
|
).prefetch_related("mentor", "course_session")
|
|
for mentor_relation in mentor_relation_qs:
|
|
cs = mentor_relation.course_session
|
|
course_session_entry = _create_course_session_dict(
|
|
cs,
|
|
"LEARNING_MENTEE",
|
|
"LEARNING_MENTOR",
|
|
)
|
|
|
|
if mentor_relation.mentor.id not in result_persons:
|
|
person_data = create_user_dict(mentor_relation.mentor)
|
|
person_data["course_sessions"] = [course_session_entry]
|
|
result_persons[mentor_relation.mentor.id] = person_data
|
|
else:
|
|
# user is already in result_persons
|
|
result_persons[mentor_relation.mentor.id]["course_sessions"].append(
|
|
course_session_entry
|
|
)
|
|
|
|
return result_persons.values()
|
|
|
|
|
|
def _persons_list_add_competence_metrics(persons):
|
|
course_session_ids = {cs["id"] for p in persons for cs in p["course_sessions"]}
|
|
competence_assignments = query_competence_course_session_assignments(
|
|
course_session_ids
|
|
) + query_competence_course_session_edoniq_tests(course_session_ids)
|
|
assignment_ids = {
|
|
a.learning_content.content_assignment.id for a in competence_assignments
|
|
}
|
|
|
|
for p in persons:
|
|
passed_count = 0
|
|
failed_count = 0
|
|
for cs in p["course_sessions"]:
|
|
evaluation_results = AssignmentCompletion.objects.filter(
|
|
completion_status=AssignmentCompletionStatus.EVALUATION_SUBMITTED.value,
|
|
assignment_user_id=p.get("user_id"),
|
|
course_session=cs.get("id"),
|
|
assignment_id__in=assignment_ids,
|
|
)
|
|
cs_passed_count = len(
|
|
[ac for ac in evaluation_results if ac.evaluation_passed]
|
|
)
|
|
cs_failed_count = len(evaluation_results) - cs_passed_count
|
|
cs["competence_metrics"] = {
|
|
"passed_count": cs_passed_count,
|
|
"failed_count": cs_failed_count,
|
|
}
|
|
passed_count += len(
|
|
[ac for ac in evaluation_results if ac.evaluation_passed]
|
|
)
|
|
failed_count += len(evaluation_results) - passed_count
|
|
p["competence_metrics"] = {
|
|
"passed_count": passed_count,
|
|
"failed_count": failed_count,
|
|
}
|
|
|
|
return persons
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_dashboard_persons(request):
|
|
try:
|
|
persons = list(_create_person_list_with_roles(request.user))
|
|
|
|
if request.GET.get("with_competence_metrics", "") == "true":
|
|
persons = _persons_list_add_competence_metrics(persons)
|
|
|
|
return Response(
|
|
status=200,
|
|
data=list(persons),
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_dashboard_due_dates(request):
|
|
try:
|
|
course_sessions = get_course_sessions_with_roles_for_user(request.user)
|
|
course_session_ids = [cs.id for cs in course_sessions]
|
|
|
|
all_due_dates = DueDate.objects.filter(
|
|
course_session__id__in=course_session_ids
|
|
)
|
|
|
|
# filter only future due dates
|
|
due_dates = []
|
|
today = date.today()
|
|
for due_date in all_due_dates:
|
|
# due_dates.append(due_date)
|
|
if due_date.end:
|
|
if due_date.end.date() >= today:
|
|
due_dates.append(due_date)
|
|
elif due_date.start:
|
|
if due_date.start.date() >= today:
|
|
due_dates.append(due_date)
|
|
|
|
due_dates.sort(key=lambda x: x.start)
|
|
|
|
# find course session by id in `course_sessions`
|
|
|
|
result_due_dates = []
|
|
for due_date in due_dates:
|
|
data = DueDateSerializer(due_date).data
|
|
|
|
cs = next(
|
|
course_session
|
|
for course_session in course_sessions
|
|
if course_session.id == due_date.course_session.id
|
|
)
|
|
if cs:
|
|
data["course_session"] = _create_course_session_dict(
|
|
cs, my_role=user_role(cs.roles), user_role=""
|
|
)
|
|
result_due_dates.append(data)
|
|
|
|
return Response(
|
|
status=200,
|
|
data=result_due_dates,
|
|
)
|
|
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
def get_widgets_for_course(
|
|
role_key: RoleKeyType, is_uk: bool, is_vv: bool, is_mentor: bool
|
|
) -> List[str]:
|
|
widgets = []
|
|
|
|
if role_key == RoleKeyType.MEMBER:
|
|
widgets.append(WidgetType.PROGRESS_WIDGET.value)
|
|
widgets.append(WidgetType.COMPETENCE_WIDGET.value)
|
|
if is_uk:
|
|
widgets.append(WidgetType.COMPETENCE_CERTIFICATE_WIDGET.value)
|
|
|
|
if role_key in [RoleKeyType.SUPERVISOR, RoleKeyType.TRAINER] and is_uk:
|
|
widgets.append(WidgetType.UK_STATISTICS_WIDGET.value)
|
|
|
|
if is_mentor:
|
|
widgets.append(WidgetType.MENTOR_PERSON_WIDGET.value)
|
|
if is_uk:
|
|
widgets.append(WidgetType.MENTOR_COMPETENCE_WIDGET.value)
|
|
if is_vv:
|
|
widgets.append(WidgetType.MENTOR_TASKS_WIDGET.value)
|
|
|
|
return widgets
|
|
|
|
|
|
def get_role_key_and_mentor(
|
|
course_sessions: List[CourseSessionWithRoles], is_uk: bool, is_vv: bool
|
|
) -> tuple[RoleKeyType, bool]:
|
|
roles = set()
|
|
role = None
|
|
for cs in course_sessions:
|
|
roles.update(cs.roles)
|
|
|
|
if "SUPERVISOR" in roles:
|
|
role = RoleKeyType.SUPERVISOR
|
|
elif "EXPERT" in roles:
|
|
role = RoleKeyType.TRAINER
|
|
elif "MEMBER" in roles:
|
|
role = RoleKeyType.MEMBER
|
|
elif "LEARNING_MENTOR" in roles:
|
|
if is_uk:
|
|
role = RoleKeyType.MENTOR_UK
|
|
elif is_vv:
|
|
role = RoleKeyType.MENTOR_VV
|
|
|
|
is_mentor = "LEARNING_MENTOR" in roles
|
|
return role, is_mentor
|
|
|
|
|
|
def collect_course_sessions_by_course(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> dict:
|
|
course_sessions_by_course = {}
|
|
for cs in course_sessions:
|
|
if cs.course.id not in course_sessions_by_course:
|
|
course_sessions_by_course[cs.course.id] = []
|
|
course_sessions_by_course[cs.course.id].append(cs)
|
|
return course_sessions_by_course
|
|
|
|
|
|
def has_preview(role_key: RoleKeyType) -> bool:
|
|
return (
|
|
role_key in [RoleKeyType.MENTOR_VV, RoleKeyType.MENTOR_UK]
|
|
and not role_key == RoleKeyType.MEMBER
|
|
)
|
|
|
|
|
|
def get_newest_cs(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> CourseSessionWithRoles | None:
|
|
newest: CourseSessionWithRoles | None = None
|
|
|
|
for cs in course_sessions:
|
|
generation_newest = newest.generation if newest else None
|
|
if generation_newest is None or cs.generation > generation_newest:
|
|
newest = cs
|
|
|
|
return newest
|
|
|
|
|
|
def get_course_config(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> List[CourseConfig]:
|
|
course_configs = []
|
|
cs_by_course = collect_course_sessions_by_course(course_sessions)
|
|
for _id, cs_in_course in cs_by_course.items():
|
|
is_uk = cs_in_course[0].course.configuration.is_uk
|
|
is_vv = cs_in_course[0].course.configuration.is_vv
|
|
role_key, is_mentor = get_role_key_and_mentor(cs_in_course, is_uk, is_vv)
|
|
session_to_continue = get_newest_cs(cs_in_course)
|
|
course_configs.append(
|
|
CourseConfig(
|
|
course_id=str(cs_in_course[0].course.id),
|
|
course_slug=cs_in_course[0].course.slug,
|
|
course_title=cs_in_course[0].course.title,
|
|
role_key=role_key.value,
|
|
is_uk=is_uk,
|
|
is_vv=is_vv,
|
|
is_mentor=is_mentor,
|
|
widgets=get_widgets_for_course(role_key, is_uk, is_vv, is_mentor),
|
|
has_preview=has_preview(role_key),
|
|
session_to_continue_id=(
|
|
str(session_to_continue.id) if session_to_continue else None
|
|
),
|
|
)
|
|
)
|
|
|
|
return course_configs
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_dashboard_config(request):
|
|
try:
|
|
course_sessions = get_course_sessions_with_roles_for_user(request.user) # noqa
|
|
course_configs = get_course_config(course_sessions)
|
|
|
|
return Response(
|
|
status=200,
|
|
data=[asdict(cc) for cc in course_configs],
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_mentee_count(request, course_id: str):
|
|
try:
|
|
return Response(
|
|
status=200,
|
|
data={"mentee_count": _get_mentee_count(course_id, request.user)}, # noqa
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
def _get_mentee_count(course_id: str, mentor: User) -> int:
|
|
return CourseSessionUser.objects.filter(
|
|
participants__mentor=mentor, course_session__course__id=course_id
|
|
).count()
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_mentor_open_tasks_count(request, course_id: str):
|
|
try:
|
|
return Response(
|
|
status=200,
|
|
data={
|
|
"open_task_count": _get_mentor_open_tasks_count(
|
|
course_id, request.user
|
|
) # noqa
|
|
},
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
|
|
def _get_mentor_open_tasks_count(course_id: str, mentor: User) -> int:
|
|
open_assigment_count = 0
|
|
open_feedback_count = 0
|
|
|
|
course_configuration = CourseConfiguration.objects.get(course_id=course_id)
|
|
|
|
if course_configuration.is_vv:
|
|
open_assigment_count = AssignmentCompletion.objects.filter(
|
|
course_session__course__id=course_id,
|
|
completion_status=AssignmentCompletionStatus.SUBMITTED.value,
|
|
evaluation_user=mentor, # noqa
|
|
assignment_user__coursesessionuser__participants__mentor=mentor,
|
|
).count()
|
|
|
|
open_feedback_qs = SelfEvaluationFeedback.objects.filter(
|
|
feedback_provider_user=mentor, # noqa
|
|
feedback_requester_user__coursesessionuser__participants__mentor=mentor,
|
|
feedback_submitted=False,
|
|
)
|
|
# filter open feedbacks for course_id (-> not possible with queryset)
|
|
open_feedback_count = len(
|
|
[
|
|
feedback_entry
|
|
for feedback_entry in open_feedback_qs
|
|
if str(feedback_entry.learning_unit.get_course().id) == course_id
|
|
]
|
|
)
|
|
|
|
return open_assigment_count + open_feedback_count
|
|
|
|
|
|
@api_view(["POST"])
|
|
def export_attendance_as_xsl(request):
|
|
circle_ids = request.data.get("circleIds", None)
|
|
requested_course_session_ids = request.data.get("courseSessionIds", [])
|
|
course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
|
|
request.user, requested_course_session_ids
|
|
)
|
|
data = export_attendance(
|
|
[cs.id for cs in course_sessions_with_roles],
|
|
circle_ids=circle_ids,
|
|
)
|
|
return _make_excel_response(data, file_name=ATTENDANCE_EXPORT_FILENAME)
|
|
|
|
|
|
@api_view(["POST"])
|
|
def export_competence_elements_as_xsl(request):
|
|
circle_ids = request.data.get("circleIds", None)
|
|
requested_course_session_ids = request.data.get("courseSessionIds", [])
|
|
course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
|
|
request.user, requested_course_session_ids
|
|
)
|
|
data = export_competence_elements(
|
|
[cswr.id for cswr in course_sessions_with_roles],
|
|
circle_ids=circle_ids,
|
|
)
|
|
return _make_excel_response(data, COMPETENCE_ELEMENT_EXPORT_FILE_NAME)
|
|
|
|
|
|
@api_view(["POST"])
|
|
def export_feedback_as_xsl(request):
|
|
circle_ids = request.data.get("circleIds", None)
|
|
requested_course_session_ids = request.data.get("courseSessionIds", [])
|
|
course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
|
|
request.user, requested_course_session_ids
|
|
) # noqa
|
|
|
|
allowed_circles = _get_permitted_circles_ids_for_user_and_course_session(
|
|
request.user,
|
|
course_sessions_with_roles,
|
|
circle_ids,
|
|
) # noqa
|
|
|
|
data = export_feedback_with_circle_restriction(allowed_circles, False)
|
|
return _make_excel_response(data, FEEDBACK_EXPORT_FILE_NAME)
|
|
|
|
|
|
def _get_permitted_courses_sessions_for_user(
|
|
user: User, requested_coursesession_ids: List[str]
|
|
) -> List[CourseSessionWithRoles]:
|
|
ALLOWED_ROLES = ["EXPERT", "SUPERVISOR"]
|
|
|
|
user_course_sessions_with_roles = _get_course_sessions_with_roles_for_user(
|
|
user, ALLOWED_ROLES, requested_coursesession_ids
|
|
) # noqa
|
|
|
|
return user_course_sessions_with_roles
|
|
|
|
|
|
def _make_excel_response(data: bytes, file_name: str) -> HttpResponse:
|
|
encoded_data = base64.b64encode(data).decode("utf-8")
|
|
|
|
# Create the JSON response
|
|
response_data = {
|
|
"encoded_data": encoded_data,
|
|
"file_name": make_export_filename(file_name),
|
|
}
|
|
|
|
return Response(response_data, status=200)
|
|
|
|
|
|
def _get_course_sessions_with_roles_for_user(
|
|
user: User, allowed_roles: List[str], requested_cs_ids: List[str]
|
|
) -> List[CourseSessionWithRoles]:
|
|
all_cs_roles_for_user = [
|
|
csr
|
|
for csr in get_course_sessions_with_roles_for_user(user)
|
|
if any(role in allowed_roles for role in csr.roles)
|
|
and csr.id in requested_cs_ids
|
|
] # noqa
|
|
|
|
return all_cs_roles_for_user
|
|
|
|
|
|
def _get_permitted_circles_ids_for_user_and_course_session(
|
|
user: User,
|
|
user_course_sessions_with_roles: List[CourseSessionWithRoles],
|
|
requested_circle_ids: Tuple[int, int],
|
|
):
|
|
allowed_circles_for_sessions = []
|
|
for cswr in user_course_sessions_with_roles:
|
|
if "SUPERVISOR" in cswr.roles:
|
|
allowed_circles_for_sessions.append((cswr.id, requested_circle_ids))
|
|
else:
|
|
course_session_users = CourseSessionUser.objects.filter(
|
|
course_session=cswr.id,
|
|
user=user,
|
|
)
|
|
allowed_circles = (
|
|
Circle.objects.filter(
|
|
Q(expert__in=course_session_users) & Q(id__in=requested_circle_ids)
|
|
)
|
|
.distinct()
|
|
.values_list("id", flat=True)
|
|
)
|
|
allowed_circles_for_sessions.append((cswr.id, list(allowed_circles)))
|
|
|
|
return allowed_circles_for_sessions
|