feat: custom roles and view-user mapping

futurex_openedx_extensions/helpers/admin.py

@@ -1,19 +1,62 @@
 """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_mysql.models import QuerySet
 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 = 'no_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', 'Yes'),
+            ('no', '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 +100,120 @@ 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__'
+
+    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']
+        ])
+
+
+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):
@@ -150,6 +304,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

futurex_openedx_extensions/helpers/constants.py

@@ -50,6 +50,13 @@
 COURSE_CREATOR_ROLE_TENANT = 'org_course_creator_group'
 COURSE_CREATOR_ROLE_GLOBAL = 'course_creator_group'
 COURSE_SUPPORT_ROLE_GLOBAL = 'support'
+COURSE_FX_API_ACCESS_ROLE = 'fx_api_access'
+COURSE_FX_API_ACCESS_ROLE_GLOBAL = 'fx_api_access_global'
+
+COURSE_ACCESS_ROLES_USER_VIEW_MAPPING = [
+    COURSE_FX_API_ACCESS_ROLE,
+    COURSE_FX_API_ACCESS_ROLE_GLOBAL
+]
 
 COURSE_ACCESS_ROLES_COURSE_ONLY = [
     'beta_testers',
@@ -68,18 +75,24 @@
     'instructor',
     COURSE_ACCESS_ROLES_LIBRARY_USER,
     COURSE_ACCESS_ROLES_STAFF_EDITOR,
+    COURSE_FX_API_ACCESS_ROLE,
 ]
 COURSE_ACCESS_ROLES_GLOBAL = [
     COURSE_CREATOR_ROLE_GLOBAL,
     COURSE_SUPPORT_ROLE_GLOBAL,
+    COURSE_FX_API_ACCESS_ROLE_GLOBAL,
 ]
 
-COURSE_ACCESS_ROLES_SUPPORTED_EDIT = \
-    COURSE_ACCESS_ROLES_COURSE_ONLY + \
-    COURSE_ACCESS_ROLES_TENANT_ONLY + \
+COURSE_ACCESS_ROLES_SUPPORTED_EDIT = list(set(
+    COURSE_ACCESS_ROLES_COURSE_ONLY +
+    COURSE_ACCESS_ROLES_TENANT_ONLY +
     COURSE_ACCESS_ROLES_TENANT_OR_COURSE
+) - {COURSE_FX_API_ACCESS_ROLE})
 
-COURSE_ACCESS_ROLES_SUPPORTED_READ = COURSE_ACCESS_ROLES_SUPPORTED_EDIT + COURSE_ACCESS_ROLES_GLOBAL
+COURSE_ACCESS_ROLES_SUPPORTED_READ = list(
+    set(COURSE_ACCESS_ROLES_SUPPORTED_EDIT + COURSE_ACCESS_ROLES_GLOBAL) |
+    {COURSE_FX_API_ACCESS_ROLE}
+)
 
 COURSE_ACCESS_ROLES_ACCEPT_COURSE_ID = COURSE_ACCESS_ROLES_COURSE_ONLY + COURSE_ACCESS_ROLES_TENANT_OR_COURSE
 

futurex_openedx_extensions/helpers/custom_roles.py

@@ -0,0 +1,75 @@
+"""New roles for FutureX Open edX Extensions."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from common.djangoapps.student.admin import CourseAccessRoleForm
+from common.djangoapps.student.roles import REGISTERED_ACCESS_ROLES, CourseRole, OrgRole, RoleBase
+
+from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
+
+log = logging.getLogger(__name__)
+
+
+def register_custom_access_role(cls: Any) -> Any:
+    """
+    Decorator that adds the new access role to the list of registered access roles to be accessible in the Django admin.
+
+    Note: roles inheritances is not supported
+
+    :param cls: The class to register
+    :type cls: Any
+    """
+    def _hacky_update_django_admin_choices(choices: list[tuple[str, str]]) -> None:
+        """
+        Update the choices of the role field in django admin.
+
+        :param choices: The choices to update
+        :type choices: list[tuple[str, str]]
+        """
+        CourseAccessRoleForm.COURSE_ACCESS_ROLES = choices
+        CourseAccessRoleForm.declared_fields['role'].choices = choices
+
+    try:
+        role_name = cls.ROLE
+        if role_name in REGISTERED_ACCESS_ROLES:
+            raise FXCodedException(
+                code=FXExceptionCodes.CUSTOM_ROLE_DUPLICATE_DECLARATION,
+                message=f'Trying to register a custom role {role_name} that is already registered!'
+            )
+    except AttributeError:
+        log.exception('Role class %s does not have a ROLE attribute', cls.__name__)
+    except FXCodedException as exc:
+        log.exception(str(exc))
+    else:
+        REGISTERED_ACCESS_ROLES[role_name] = cls
+        _hacky_update_django_admin_choices([(role.ROLE, role.ROLE) for role in REGISTERED_ACCESS_ROLES.values()])
+
+    return cls
+
+
+@register_custom_access_role
+class FXAPIAccessRoleCourse(CourseRole):  # pylint: disable=too-few-public-methods
+    """Course specific access to the FutureX APIs."""
+    ROLE = 'fx_api_access'
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(self.ROLE, *args, **kwargs)
+
+
+class FXAPIAccessRoleOrg(OrgRole):  # pylint: disable=too-few-public-methods
+    """Tenant-wide access to the FutureX APIs."""
+    ROLE = 'fx_api_access'
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(self.ROLE, *args, **kwargs)
+
+
+@register_custom_access_role
+class FXAPIAccessRoleGlobal(RoleBase):  # pylint: disable=too-few-public-methods
+    """Global access to the FutureX APIs."""
+    ROLE = 'fx_api_access_global'
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(self.ROLE, *args, **kwargs)

futurex_openedx_extensions/helpers/exceptions.py

@@ -40,6 +40,8 @@ class FXExceptionCodes(Enum):
 
     QUERY_SET_BAD_OPERATION = 8001
 
+    CUSTOM_ROLE_DUPLICATE_DECLARATION = 9001
+
 
 class FXCodedException(Exception):
     """Exception with a code."""

futurex_openedx_extensions/helpers/migrations/0005_view_user_mapping.py

@@ -0,0 +1,53 @@
+# Generated by Django 3.2.25 on 2024-11-19 12:09
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import simple_history.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('fx_helpers', '0004_dataexporttask'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='HistoricalViewUserMapping',
+            fields=[
+                ('id', models.IntegerField(auto_created=True, blank=True, db_index=True, verbose_name='ID')),
+                ('view_name', models.CharField(max_length=255)),
+                ('enabled', models.BooleanField(default=True)),
+                ('expires_at', models.DateTimeField(blank=True, null=True)),
+                ('history_id', models.AutoField(primary_key=True, serialize=False)),
+                ('history_date', models.DateTimeField()),
+                ('history_change_reason', models.CharField(max_length=100, null=True)),
+                ('history_type', models.CharField(choices=[('+', 'Created'), ('~', 'Changed'), ('-', 'Deleted')], max_length=1)),
+                ('history_user', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
+                ('user', models.ForeignKey(blank=True, db_constraint=False, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='+', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'historical View-User Mapping',
+                'ordering': ('-history_date', '-history_id'),
+                'get_latest_by': 'history_date',
+            },
+            bases=(simple_history.models.HistoricalChanges, models.Model),
+        ),
+        migrations.CreateModel(
+            name='ViewUserMapping',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('view_name', models.CharField(max_length=255)),
+                ('enabled', models.BooleanField(default=True)),
+                ('expires_at', models.DateTimeField(blank=True, null=True)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'View-User Mapping',
+                'verbose_name_plural': 'Views-Users Mapping',
+                'unique_together': {('user', 'view_name')},
+            },
+        ),
+    ]