feat: enable OAuth2 per View

futurex_openedx_extensions/__init__.py

@@ -1,3 +1,3 @@
 """One-line description for README and other doc files."""
 
-__version__ = '0.9.14'
+__version__ = '0.9.15'

futurex_openedx_extensions/dashboard/views.py

@@ -43,6 +43,7 @@
     COURSE_ACCESS_ROLES_SUPPORTED_READ,
     COURSE_STATUS_SELF_PREFIX,
     COURSE_STATUSES,
+    FX_VIEW_DEFAULT_AUTH_CLASSES,
 )
 from futurex_openedx_extensions.helpers.converters import error_details_to_dictionary
 from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
@@ -69,6 +70,8 @@
 from futurex_openedx_extensions.helpers.tenants import get_tenants_info
 from futurex_openedx_extensions.helpers.users import get_user_by_key
 
+default_auth_classes = FX_VIEW_DEFAULT_AUTH_CLASSES.copy()
+
 
 class TotalCountsView(APIView, FXViewRoleInfoMixin):
     """
@@ -91,6 +94,7 @@ class TotalCountsView(APIView, FXViewRoleInfoMixin):
         STAT_LEARNERS: 'learners_count',
     }
 
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'total_counts_statistics'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -182,8 +186,9 @@ def get(self, request: Any, *args: Any, **kwargs: Any) -> Response | JsonRespons
 
 class LearnersView(ListAPIView, FXViewRoleInfoMixin):
     """View to get the list of learners"""
-    serializer_class = serializers.LearnerDetailsSerializer
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
+    serializer_class = serializers.LearnerDetailsSerializer
     pagination_class = DefaultPagination
     fx_view_name = 'learners_list'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -203,8 +208,9 @@ def get_queryset(self) -> QuerySet:
 
 class CoursesView(ListAPIView, FXViewRoleInfoMixin):
     """View to get the list of courses"""
-    serializer_class = serializers.CourseDetailsSerializer
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
+    serializer_class = serializers.CourseDetailsSerializer
     pagination_class = DefaultPagination
     filter_backends = [DefaultOrderingFilter]
     ordering_fields = [
@@ -231,6 +237,7 @@ def get_queryset(self) -> QuerySet:
 
 class CourseStatusesView(APIView, FXViewRoleInfoMixin):
     """View to get the course statuses"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'course_statuses'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -264,6 +271,7 @@ def get(self, request: Any, *args: Any, **kwargs: Any) -> JsonResponse:
 
 class LearnerInfoView(APIView, FXViewRoleInfoMixin):
     """View to get the information of a learner"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'learner_detailed_info'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -298,8 +306,9 @@ def get(self, request: Any, username: str, *args: Any, **kwargs: Any) -> JsonRes
 
 class DataExportManagementView(viewsets.ModelViewSet, FXViewRoleInfoMixin):  # pylint: disable=too-many-ancestors
     """View to list and retrieve data export tasks."""
-    serializer_class = serializers.DataExportTaskSerializer
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
+    serializer_class = serializers.DataExportTaskSerializer
     pagination_class = DefaultPagination
     fx_view_name = 'exported_files_data'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -328,6 +337,7 @@ def get_object(self) -> DataExportTask:
 
 class LearnerCoursesView(APIView, FXViewRoleInfoMixin):
     """View to get the list of courses for a learner"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     pagination_class = DefaultPagination
     fx_view_name = 'learner_courses'
@@ -399,6 +409,7 @@ def get(self, request: Any, *args: Any, **kwargs: Any) -> JsonResponse:  # pylin
 
 class LearnersDetailsForCourseView(ExportCSVMixin, ListAPIView, FXViewRoleInfoMixin):
     """View to get the list of learners for a course"""
+    authentication_classes = default_auth_classes
     serializer_class = serializers.LearnerDetailsForCourseSerializer
     permission_classes = [FXHasTenantCourseAccess]
     pagination_class = DefaultPagination
@@ -467,6 +478,7 @@ def get_serializer_context(self) -> Dict[str, Any]:
 
 class GlobalRatingView(APIView, FXViewRoleInfoMixin):
     """View to get the global rating"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'global_rating'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
@@ -494,6 +506,7 @@ def get(self, request: Any, *args: Any, **kwargs: Any) -> JsonResponse:
 
 class UserRolesManagementView(viewsets.ModelViewSet, FXViewRoleInfoMixin):  # pylint: disable=too-many-ancestors
     """View to get the user roles"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantAllCoursesAccess]
     fx_view_name = 'user_roles'
     fx_default_read_only_roles = ['org_course_creator_group']
@@ -656,6 +669,7 @@ def destroy(self, request: Any, *args: Any, **kwargs: Any) -> Response:
 
 class MyRolesView(APIView, FXViewRoleInfoMixin):
     """View to get the user roles of the caller"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'my_roles'
     fx_default_read_only_roles = COURSE_ACCESS_ROLES_SUPPORTED_READ.copy()
@@ -672,6 +686,7 @@ def get(self, request: Any, *args: Any, **kwargs: Any) -> JsonResponse:
 
 class ClickhouseQueryView(APIView, FXViewRoleInfoMixin):
     """View to get the Clickhouse query"""
+    authentication_classes = default_auth_classes
     permission_classes = [FXHasTenantCourseAccess]
     fx_view_name = 'clickhouse_query_fetcher'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']

futurex_openedx_extensions/helpers/admin.py

@@ -1,19 +1,64 @@
 """Django admin view for the models."""
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, List, Tuple
 
 import yaml  # type: ignore
+from common.djangoapps.student.admin import CourseAccessRoleForm
+from django import forms
 from django.contrib import admin
+from django.contrib.admin import SimpleListFilter
 from django.core.cache import cache
 from django.http import Http404, HttpResponseRedirect
 from django.urls import path
 from django.utils import timezone
+from django.utils.translation.trans_null import gettext_lazy
+from django_mysql.models import QuerySet
+from openedx.core.lib.api.authentication import BearerAuthentication
 from rest_framework.response import Response
 from simple_history.admin import SimpleHistoryAdmin
 
 from futurex_openedx_extensions.helpers.constants import CACHE_NAMES
-from futurex_openedx_extensions.helpers.models import ClickhouseQuery, DataExportTask, ViewAllowedRoles
+from futurex_openedx_extensions.helpers.models import ClickhouseQuery, DataExportTask, ViewAllowedRoles, ViewUserMapping
+from futurex_openedx_extensions.helpers.roles import get_fx_view_with_roles
+
+
+class YesNoFilter(SimpleListFilter):
+    """Filter for the Yes/No fields."""
+    title = 'Yes / No'
+    parameter_name = 'not_yet_set_must_be_replaced'
+
+    def lookups(self, request: Any, model_admin: Any) -> List[Tuple[str, str]]:
+        """
+        Define filter options.
+
+        :param request: The request object
+        :type request: Request
+        :param model_admin: The model admin object
+        :type model_admin: Any
+        :return: List of filter options
+        """
+        return [
+            ('yes', gettext_lazy('Yes')),
+            ('no', gettext_lazy('No')),
+        ]
+
+    def queryset(self, request: Any, queryset: QuerySet) -> QuerySet:
+        """
+        Filter the queryset based on the selected option.
+
+        :param request: The request object
+        :type request: Request
+        :param queryset: The queryset to filter
+        :type queryset: QuerySet
+        :return: The filtered queryset
+        """
+        filter_params = {self.parameter_name: self.value() == 'yes'}
+
+        if self.value() in ('yes', 'no'):
+            return queryset.filter(**filter_params)
+
+        return queryset
 
 
 class ClickhouseQueryAdmin(SimpleHistoryAdmin):
@@ -57,9 +102,132 @@ def load_missing_queries(self, request: Any) -> HttpResponseRedirect:  # pylint:
     list_display = ('id', 'scope', 'version', 'slug', 'description', 'enabled', 'modified_at')
 
 
+class ViewAllowedRolesModelForm(forms.ModelForm):
+    """Model form for the ViewAllowedRoles model."""
+    class Meta:
+        """Meta class for the ViewAllowedRoles model form."""
+        model = ViewAllowedRoles
+        fields = '__all__'
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Initialize the form with the dynamic choices."""
+        super().__init__(*args, **kwargs)
+
+        self.fields['view_name'] = forms.TypedChoiceField()
+        self.fields['view_name'].choices = sorted([
+            (view_name, view_name) for view_name in get_fx_view_with_roles()['_all_view_names']
+        ])
+
+        self.fields['allowed_role'] = forms.TypedChoiceField()
+        self.fields['allowed_role'].choices = CourseAccessRoleForm.COURSE_ACCESS_ROLES
+
+
 class ViewAllowedRolesHistoryAdmin(SimpleHistoryAdmin):
     """Admin view for the ViewAllowedRoles model."""
+    form = ViewAllowedRolesModelForm
+
     list_display = ('view_name', 'view_description', 'allowed_role')
+    list_filter = ('view_name', 'allowed_role')
+
+
+class IsUserActiveFilter(YesNoFilter):
+    """Filter for the is_user_active field."""
+    title = 'Is User Active'
+    parameter_name = 'is_user_active'
+
+
+class IsUserSystemStaffFilter(YesNoFilter):
+    """Filter for the is_user_system_staff field."""
+    title = 'Is User System Staff'
+    parameter_name = 'is_user_system_staff'
+
+
+class UsableFilter(YesNoFilter):
+    """Filter for the usable field."""
+    title = 'Usable'
+    parameter_name = 'usable'
+
+
+class HasAccessRomeFilter(YesNoFilter):
+    """Filter for the usable field."""
+    title = 'Has Access Role'
+    parameter_name = 'has_access_role'
+
+
+class ViewUserMappingModelForm(forms.ModelForm):
+    """Model form for the ViewUserMapping model."""
+    class Meta:
+        """Meta class for the ViewUserMapping model form."""
+        model = ViewUserMapping
+        fields = '__all__'
+
+    @staticmethod
+    def get_all_supported_view_names() -> List[Any]:
+        """Get all the supported view names."""
+        result = []
+
+        for view_name, view_class in get_fx_view_with_roles()['_all_view_names'].items():
+            if hasattr(
+                view_class, 'authentication_classes',
+            ) and BearerAuthentication in view_class.authentication_classes:
+                result.append(view_name)
+
+        return sorted(result)
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Initialize the form with the dynamic choices."""
+        super().__init__(*args, **kwargs)
+
+        self.fields['view_name'] = forms.TypedChoiceField()
+        self.fields['view_name'].choices = [(view_name, view_name) for view_name in self.get_all_supported_view_names()]
+
+
+class ViewUserMappingHistoryAdmin(SimpleHistoryAdmin):
+    """Admin view for the ViewUserMapping model."""
+    form = ViewUserMappingModelForm
+
+    list_display = (
+        'user', 'view_name', 'enabled', 'expires_at', 'is_user_active',
+        'is_user_system_staff', 'has_access_role', 'usable',
+    )
+    list_filter = (
+        'view_name', 'enabled', 'expires_at',
+        IsUserActiveFilter, IsUserSystemStaffFilter, HasAccessRomeFilter, UsableFilter,
+    )
+    search_fields = ('user__username', 'user__email')
+    raw_id_fields = ('user',)
+
+    def is_user_active(self, obj: ViewUserMapping) -> bool:  # pylint: disable=no-self-use
+        """Check if the user is active or not."""
+        return obj.get_is_user_active()
+
+    def is_user_system_staff(self, obj: ViewUserMapping) -> bool:  # pylint: disable=no-self-use
+        """Check if the user is system staff or not."""
+        return obj.get_is_user_system_staff()
+
+    def has_access_role(self, obj: ViewUserMapping) -> bool:  # pylint: disable=no-self-use
+        """Check if the user has access role."""
+        return obj.get_has_access_role()
+
+    def usable(self, obj: ViewUserMapping) -> bool:  # pylint: disable=no-self-use
+        """Check if the mapping link is usable."""
+        return obj.get_usable()
+
+    is_user_active.short_description = 'Is User Active'  # type: ignore
+    is_user_active.boolean = True  # type: ignore
+    is_user_active.admin_order_field = 'is_user_active'  # type: ignore
+
+    is_user_system_staff.short_description = 'Is User System Staff'  # type: ignore
+    is_user_system_staff.boolean = True  # type: ignore
+    is_user_system_staff.admin_order_field = 'is_user_system_staff'  # type: ignore
+
+    has_access_role.short_description = 'Has Access Role'  # type: ignore
+    has_access_role.boolean = True  # type: ignore
+    has_access_role.admin_order_field = 'has_access_role'  # type: ignore
+
+    usable.short_description = 'Usable'  # type: ignore
+    usable.boolean = True  # type: ignore
+    usable.admin_order_field = 'usable'  # type: ignore
 
 
 class CacheInvalidator(ViewAllowedRoles):
@@ -151,6 +319,7 @@ def register_admins() -> None:
     admin.site.register(CacheInvalidator, CacheInvalidatorAdmin)
     admin.site.register(ClickhouseQuery, ClickhouseQueryAdmin)
     admin.site.register(ViewAllowedRoles, ViewAllowedRolesHistoryAdmin)
+    admin.site.register(ViewUserMapping, ViewUserMappingHistoryAdmin)
     admin.site.register(DataExportTask, DataExportTaskAdmin)
 
 

futurex_openedx_extensions/helpers/apps.py

@@ -29,6 +29,8 @@ class HelpersConfig(AppConfig):
 
     def ready(self) -> None:
         """Connect handlers to send notifications about discussions."""
+        from futurex_openedx_extensions.helpers import \
+            custom_roles  # pylint: disable=unused-import, import-outside-toplevel
         from futurex_openedx_extensions.helpers import \
             monkey_patches  # pylint: disable=unused-import, import-outside-toplevel
         from futurex_openedx_extensions.helpers import signals  # pylint: disable=unused-import, import-outside-toplevel