vbv/server/vbv_lernwelt/dashboard/graphql/queries.py

from typing import Dict, List, Set, Tuple

import graphene

from vbv_lernwelt.core.admin import User
from vbv_lernwelt.course.models import Course, CourseSession, CourseSessionUser
from vbv_lernwelt.course_session_group.models import CourseSessionGroup
from vbv_lernwelt.dashboard.graphql.types.dashboard import (
    CourseProgressType,
    CourseStatisticsType,
    DashboardConfigType,
    DashboardType,
)
from vbv_lernwelt.iam.permissions import (
    can_view_course_session,
    can_view_course_session_group_statistics,
    can_view_course_session_progress,
)


class DashboardQuery(graphene.ObjectType):
    course_statistics = graphene.Field(
        CourseStatisticsType, course_id=graphene.ID(required=True)
    )

    course_progress = graphene.Field(
        CourseProgressType, course_id=graphene.ID(required=True)
    )

    dashboard_config = graphene.List(
        graphene.NonNull(DashboardConfigType), required=True
    )

    def resolve_course_statistics(root, info, course_id: str):  # noqa
        user = info.context.user
        course = Course.objects.get(id=course_id)

        course_session_ids = set()

        for group in CourseSessionGroup.objects.filter(course=course):
            if can_view_course_session_group_statistics(user=user, group=group):
                course_session_ids.update(
                    group.course_session.all().values_list("id", flat=True)
                )

        if not course_session_ids:
            return None

        return CourseStatisticsType(
            id=course.id,  # noqa
            course_title=course.title,  # noqa
            course_session_selection_ids=list(course_session_ids),  # noqa
        )

    def resolve_dashboard_config(root, info):  # noqa
        user = info.context.user

        if user.is_superuser:
            courses = Course.objects.all().values("id", "title", "slug")
            return [
                {
                    "id": c["id"],
                    "name": c["title"],
                    "slug": c["slug"],
                    "dashboard_type": DashboardType.SIMPLE_LIST_DASHBOARD,
                }
                for c in courses
            ]

        (
            statistic_dashboards,
            statistics_dashboard_course_ids,
        ) = get_user_statistics_dashboards(user=user)

        course_session_dashboards = get_user_course_session_dashboards(
            user=user, exclude_course_ids=statistics_dashboard_course_ids
        )

        return statistic_dashboards + course_session_dashboards

    def resolve_course_progress(root, info, course_id: str):  # noqa
        """
        Slightly fragile but could be good enough: most only have one
        course session per course anyway but if there are multiple, we
        just pick the newest one (by generation) as best guess.
        """

        user = info.context.user
        newest: CourseSession | None = None

        for course_session in CourseSession.objects.filter(course_id=course_id):
            if can_view_course_session_progress(
                    user=user, course_session=course_session
            ):
                generation_newest = newest.generation if newest else None
                if (
                        generation_newest is None
                        or course_session.generation > generation_newest
                ):
                    newest = course_session

        if not newest:
            return None

        return CourseProgressType(
            id=course_id, session_to_continue_id=newest.id  # noqa  # noqa
        )


def get_user_statistics_dashboards(user: User) -> Tuple[List[Dict[str, str]], Set[int]]:
    course_index = set()
    dashboards = []

    for group in CourseSessionGroup.objects.all():
        if can_view_course_session_group_statistics(user=user, group=group):
            course = group.course
            course_index.add(course)
            dashboards.append(
                {
                    "id": str(course.id),
                    "name": course.title,
                    "slug": course.slug,
                    "dashboard_type": DashboardType.STATISTICS_DASHBOARD,
                }
            )

    return dashboards, course_index


def get_user_course_session_dashboards(
        user: User, exclude_course_ids: Set[int]
) -> List[Dict[str, str]]:
    """
    Edge case: what do we show to users with access to multiple
               sessions of a course, but with varying permissions?
               -> We just show the simple list dashboard for now.
    """

    dashboards = []

    course_sessions = CourseSession.objects.exclude(course__in=exclude_course_ids)
    roles_by_course: Dict[Course, Set[DashboardType]] = {}

    for course_session in course_sessions:
        if can_view_course_session(user=user, course_session=course_session):
            role = CourseSessionUser.objects.get(
                course_session=course_session, user=user
            ).role
            roles_by_course.setdefault(course_session.course, set())
            roles_by_course[course_session.course].add(role)

    for course, roles in roles_by_course.items():
        resolved_dashboard_type = None

        if len(roles) == 1:
            course_role = roles.pop()
            if course_role == CourseSessionUser.Role.EXPERT:
                resolved_dashboard_type = DashboardType.SIMPLE_LIST_DASHBOARD
            elif course_role == CourseSessionUser.Role.MEMBER:
                resolved_dashboard_type = DashboardType.PROGRESS_DASHBOARD
        else:
            # fallback: just go with simple list dashboard
            resolved_dashboard_type = DashboardType.SIMPLE_LIST_DASHBOARD

        dashboards.append(
            {
                "id": str(course.id),
                "name": course.title,
                "slug": course.slug,
                "dashboard_type": resolved_dashboard_type,
            }
        )

    return dashboards