from dataclasses import dataclass from typing import List, Set from vbv_lernwelt.core.models import User from vbv_lernwelt.course.models import CourseSession, CourseSessionUser from vbv_lernwelt.course_session_group.models import CourseSessionGroup from vbv_lernwelt.learning_mentor.models import ( AgentParticipantRelation, AgentParticipantRoleType, ) @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.") 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 agent_qs = AgentParticipantRelation.objects.filter(agent=user).prefetch_related( "participant__course_session", "participant__course_session__course" ) for agent_relation in agent_qs: cs = agent_relation.participant.course_session cs.roles = set() cs = result_course_sessions.get(cs.id, cs) cs.roles.add(agent_relation.role) 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, course_session_ids=None, include_private_data=False ): def create_csu_dict(csu: CourseSessionUser): user_data = create_user_dict(csu.user) user_data["chosen_profile"] = ( csu.chosen_profile.code if csu.chosen_profile else "all" ) user_data["paid_datetime"] = ( csu.paid_datetime.isoformat() if csu.paid_datetime else None ) return user_data def create_user_dict(user_object: User): user_data = { "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": [], } if include_private_data: user_data["phone_number"] = user_object.phone_number user_data["Lehrvertragsnummer"] = user_object.additional_json_data.get( "Lehrvertragsnummer", "" ) return user_data course_sessions = get_course_sessions_with_roles_for_user(user) if course_session_ids: course_sessions = [ csr for csr in course_sessions if csr._original.id in course_session_ids ] 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 ).select_related("user") my_role = user_role(cs.roles) for csu in course_session_users: person_data = result_persons.get(csu.user.id, create_csu_dict(csu)) 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: def _add_agent_relation(cs, relation, my_role, user_role): course_session_entry = create_course_session_dict(cs, my_role, user_role) participant_user = relation.participant.user if participant_user.id not in result_persons: person_data = create_csu_dict(relation.participant) 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 ) if "LEARNING_MENTOR" in cs.roles: for relation in AgentParticipantRelation.objects.filter( agent=user, participant__course_session_id=cs.id, role="LEARNING_MENTOR", ): _add_agent_relation(cs, relation, "LEARNING_MENTOR", "PARTICIPANT") if "BERUFSBILDNER" in cs.roles: for relation in AgentParticipantRelation.objects.filter( agent=user, participant__course_session_id=cs.id, role="BERUFSBILDNER", ): _add_agent_relation(cs, relation, "BERUFSBILDNER", "PARTICIPANT") # add persons where request.user is lerning mentee mentor_relation_qs = AgentParticipantRelation.objects.filter( participant__user=user, role=AgentParticipantRoleType.LEARNING_MENTOR.value, ).prefetch_related("agent") for mentor_relation in mentor_relation_qs: cs = mentor_relation.participant.course_session course_session_entry = create_course_session_dict( cs, "PARTICIPANT", "LEARNING_MENTOR", ) if mentor_relation.agent.id not in result_persons: person_data = create_user_dict(mentor_relation.agent) person_data["course_sessions"] = [course_session_entry] result_persons[mentor_relation.agent.id] = person_data else: # user is already in result_persons result_persons[mentor_relation.agent.id]["course_sessions"].append( course_session_entry ) return result_persons.values()