feat: custom roles and view-user mapping

futurex_openedx_extensions/helpers/admin.py

@@ -5,15 +5,28 @@
 
 import yaml  # type: ignore
 from django.contrib import admin
+from django.contrib.admin import SimpleListFilter
 from django.core.cache import cache
+from django.db.models import Q
 from django.http import Http404, HttpResponseRedirect
 from django.urls import path
 from django.utils import timezone
+from django.utils.html import format_html
 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
+
+
+def get_field_span(value: str) -> str:
+    """Get the field span HTML."""
+    defined = {
+        'Yes': '<span style="color: green;">Yes</span>',
+        'No': '<span style="color: red;">No</span>',
+        '?': f'<span style="color: blue;">?</span>',
+    }
+    return format_html(defined.get(value, defined['?']))
 
 
 class ClickhouseQueryAdmin(SimpleHistoryAdmin):
@@ -62,6 +75,64 @@ class ViewAllowedRolesHistoryAdmin(SimpleHistoryAdmin):
     list_display = ('view_name', 'view_description', 'allowed_role')
 
 
+class UsableFilter(SimpleListFilter):
+    title = 'Usable'
+    parameter_name = 'usable'
+
+    def lookups(self, request, model_admin):
+        """Define filter options."""
+        return (
+            ('yes', 'Yes'),
+            ('no', 'No'),
+        )
+
+    def queryset(self, request, queryset):
+        """Filter the queryset based on the selected option."""
+        if self.value() == 'yes':
+            return queryset.filter(
+                user__is_active=True, enabled=True,
+            ).filter(
+                Q(expires_at__isnull=True) | Q(expires_at__gt=timezone.now()),
+            )
+        if self.value() == 'no':
+            return queryset.filter(
+                Q(user__is_active=False) |
+                Q(enabled=False) |
+                (Q(expires_at__isnull=False) | Q(expires_at__lt=timezone.now())),
+            )
+
+        return queryset
+
+
+class ViewUserMappingHistoryAdmin(SimpleHistoryAdmin):
+    """Admin view for the ViewUserMapping model."""
+    list_display = ('user', 'view_name', 'enabled', 'expires_at', 'user_active', 'has_access_role', 'usable')
+    list_filter = ('view_name', 'enabled', 'expires_at', UsableFilter)
+    search_fields = ('user__username', 'user__email')
+
+    def user_active(self, obj):
+        """Check if the user is active or not."""
+        if obj.user.is_active:
+            return get_field_span('Yes')
+        return get_field_span('No')
+
+    def usable(self, obj):
+        """Check if the mapping link is usable."""
+        if obj.usable:
+            return get_field_span('Yes')
+        return get_field_span('No')
+
+    def has_access_role(self, obj):
+        """Check if the user has access role."""
+        if obj.usable:
+            return get_field_span('Yes')
+        return get_field_span('No')
+
+    user_active.short_description = 'User is Active'
+    has_access_role.short_description = 'Has Access Role'
+    usable.short_description = 'Usable'
+
+
 class CacheInvalidator(ViewAllowedRoles):
     """Dummy class to be able to register the Non-Model admin view CacheInvalidatorAdmin."""
     class Meta:
@@ -150,6 +221,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,4 +29,5 @@ 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 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,63 @@
+"""New roles for FutureX Open edX Extensions."""
+import logging
+
+from common.djangoapps.student.admin import CourseAccessRoleForm
+from common.djangoapps.student.roles import CourseRole, OrgRole, REGISTERED_ACCESS_ROLES, RoleBase
+
+from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
+from futurex_openedx_extensions.helpers.models import ViewAllowedRoles
+
+
+log = logging.getLogger(__name__)
+
+
+def register_custom_access_role(cls):
+    """
+    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
+    """
+    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)
+    except FXCodedException as exc:
+        log.exception(str(exc))
+    else:
+        REGISTERED_ACCESS_ROLES[role_name] = cls
+        CourseAccessRoleForm.COURSE_ACCESS_ROLES.append((cls.ROLE, cls.ROLE))
+        CourseAccessRoleForm.declared_fields['role'].choices = CourseAccessRoleForm.COURSE_ACCESS_ROLES
+        ViewAllowedRoles.allowed_role.field.choices = CourseAccessRoleForm.COURSE_ACCESS_ROLES
+
+    return cls
+
+
+@register_custom_access_role
+class FXAPIAccessRoleCourse(CourseRole):
+    """Course specific access to the FutureX APIs."""
+    ROLE = 'fx_api_access'
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(self.ROLE, *args, **kwargs)
+
+
+class FXAPIAccessRoleOrg(OrgRole):
+    """Tenant-wide access to the FutureX APIs."""
+    ROLE = 'fx_api_access'
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(self.ROLE, *args, **kwargs)
+
+
+@register_custom_access_role
+class FXAPIAccessRoleGlobal(RoleBase):
+    """Global access to the FutureX APIs."""
+    ROLE = 'fx_api_access_global'
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(self.ROLE, *args, **kwargs)

futurex_openedx_extensions/helpers/exceptions.py

@@ -38,6 +38,8 @@ class FXExceptionCodes(Enum):
 
     SERIALIZER_FILED_NAME_DOES_NOT_EXIST = 7001
 
+    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,68 @@
+# Generated by Django 3.2.25 on 2024-11-19 05:51
+
+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.AlterField(
+            model_name='historicalviewallowedroles',
+            name='allowed_role',
+            field=models.CharField(choices=[('fx_api_access', 'fx_api_access'), ('fx_api_access_global', 'fx_api_access_global')], max_length=255),
+        ),
+        migrations.AlterField(
+            model_name='viewallowedroles',
+            name='allowed_role',
+            field=models.CharField(choices=[('fx_api_access', 'fx_api_access'), ('fx_api_access_global', 'fx_api_access_global')], max_length=255),
+        ),
+        migrations.AlterField(
+            model_name='viewallowedroles',
+            name='view_name',
+            field=models.CharField(choices=[('clickhouse_query_fetcher', 'clickhouse_query_fetcher'), ('course_statuses', 'course_statuses'), ('courses_list', 'courses_list'), ('exported_files_data', 'exported_files_data'), ('global_rating', 'global_rating'), ('learner_courses', 'learner_courses'), ('learner_detailed_info', 'learner_detailed_info'), ('learners_list', 'learners_list'), ('learners_with_details_for_course', 'learners_with_details_for_course'), ('my_roles', 'my_roles'), ('total_counts_statistics', 'total_counts_statistics'), ('user_roles', 'user_roles')], max_length=255),
+        ),
+        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(choices=[('clickhouse_query_fetcher', 'clickhouse_query_fetcher'), ('course_statuses', 'course_statuses'), ('courses_list', 'courses_list'), ('exported_files_data', 'exported_files_data'), ('global_rating', 'global_rating'), ('learner_courses', 'learner_courses'), ('learner_detailed_info', 'learner_detailed_info'), ('learners_list', 'learners_list'), ('learners_with_details_for_course', 'learners_with_details_for_course'), ('my_roles', 'my_roles'), ('total_counts_statistics', 'total_counts_statistics'), ('user_roles', 'user_roles')], 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': {('view_name', 'user')},
+            },
+        ),
+    ]

futurex_openedx_extensions/helpers/models.py

@@ -4,15 +4,21 @@
 import re
 from typing import Any, Dict, List, Tuple
 
+from common.djangoapps.student.admin import CourseAccessRoleForm
+from common.djangoapps.student.models import CourseAccessRole
 from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db import models
+from django.db.models import Q
 from django.utils import timezone
 from eox_tenant.models import TenantConfig
+from opaque_keys.edx.django.models import CourseKeyField
 from simple_history.models import HistoricalRecords
 
 from futurex_openedx_extensions.helpers import clickhouse_operations as ch
+from futurex_openedx_extensions.helpers import constants as cs
 from futurex_openedx_extensions.helpers.converters import DateMethods
+from futurex_openedx_extensions.helpers.converters import get_allowed_roles
 from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
 
 User = get_user_model()
@@ -22,7 +28,7 @@ class ViewAllowedRoles(models.Model):
     """Allowed roles for every supported view"""
     view_name = models.CharField(max_length=255)
     view_description = models.CharField(null=True, blank=True, max_length=255)
-    allowed_role = models.CharField(max_length=255)
+    allowed_role = models.CharField(max_length=255, choices=CourseAccessRoleForm.COURSE_ACCESS_ROLES)
     allow_write = models.BooleanField(default=False)
 
     history = HistoricalRecords()
@@ -34,6 +40,70 @@ class Meta:
         unique_together = ('view_name', 'allowed_role')
 
 
+class ViewUserMapping(models.Model):
+    """Allowed roles for every supported view"""
+    user = models.ForeignKey(User, on_delete=models.CASCADE)
+    view_name = models.CharField(max_length=255)
+    enabled = models.BooleanField(default=True)
+    expires_at = models.DateTimeField(null=True, blank=True)
+
+    history = HistoricalRecords()
+
+    @property
+    def usable(self):
+        """Check if the mapping is usable"""
+        allowed = self.enabled and self.user.is_active and (self.expires_at is None or self.expires_at > timezone.now())
+        return allowed and self.has_access_role
+
+    @property
+    def has_access_role(self):
+        """Check if the user has at least one of view-user-mapping access roles"""
+        allowed_roles = get_allowed_roles(cs.COURSE_ACCESS_ROLES_USER_VIEW_MAPPING)
+        return CourseAccessRole.objects.filter(
+            user=self.user,
+        ).filter(
+            (
+                Q(role__in=allowed_roles['global']) &
+                Q(org='') &
+                Q(course_id=CourseKeyField.Empty)
+            ) |
+            (
+                Q(role__in=allowed_roles['tenant_only']) &
+                ~Q(org='')
+            ) |
+            (
+                Q(role__in=allowed_roles['course_only']) &
+                ~Q(org='') &
+                ~Q(course_id=CourseKeyField.Empty)
+            ) |
+            (
+                Q(role__in=allowed_roles['tenant_or_course']) &
+                ~Q(org='')
+            ),
+        ).exists()
+
+    @classmethod
+    def is_usable_access(cls, user: User, view_name: str) -> bool:
+        """
+        Check if the user has usable access to the view.
+
+        :param user: The user to check.
+        :type user: User
+        :param view_name: The name of the view.
+        :type view_name: str
+        :return: True if the user has usable access to the view.
+        :rtype: bool
+        """
+        record = cls.objects.filter(user=user, view_name=view_name).first()
+        return record is not None and record.usable
+
+    class Meta:
+        """Metaclass for the model"""
+        verbose_name = 'View-User Mapping'
+        verbose_name_plural = 'Views-Users Mapping'
+        unique_together = ('view_name', 'user')
+
+
 class ClickhouseQuery(models.Model):
     """Model for storing Clickhouse queries"""
     SCOPE_COURSE = 'course'