refactor: address quality issues

Author: mariajgrimaldi

openedx_authz/api/data.py

@@ -7,7 +7,6 @@
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.locator import LibraryLocatorV2
 
-
 AUTHZ_POLICY_ATTRIBUTES_SEPARATOR = "^"
 
 
@@ -77,14 +76,14 @@ def __attrs_post_init__(self):
 class ScopeMeta(type):
     """Metaclass for ScopeData to handle dynamic subclass instantiation based on namespace."""
 
-    _scope_registry: ClassVar[dict[str, Type["ScopeData"]]] = {}
+    scope_registry: ClassVar[dict[str, Type["ScopeData"]]] = {}
 
     def __init__(cls, name, bases, attrs):
         """Initialize the metaclass and register subclasses."""
         super().__init__(name, bases, attrs)
-        if not hasattr(cls, "_scope_registry"):
-            cls._scope_registry = {}
-        cls._scope_registry[cls.NAMESPACE] = cls
+        if not hasattr(cls, "scope_registry"):
+            cls.scope_registry = {}
+        cls.scope_registry[cls.NAMESPACE] = cls
 
     def __call__(cls, *args, **kwargs):
         """Instantiate the appropriate subclass based on the namespace in namespaced_key.
@@ -115,7 +114,8 @@ def __call__(cls, *args, **kwargs):
 
         return super().__call__(*args, **kwargs)
 
-    def get_subclass_by_namespaced_key(cls, namespaced_key: str) -> Type["ScopeData"]:
+    @classmethod
+    def get_subclass_by_namespaced_key(mcs, namespaced_key: str) -> Type["ScopeData"]:
         """Get the appropriate subclass based on the namespace in namespaced_key.
 
         Args:
@@ -126,9 +126,10 @@ def get_subclass_by_namespaced_key(cls, namespaced_key: str) -> Type["ScopeData"
         """
         # TODO: Default separator, can't access directly from class so made it a constant
         namespace = namespaced_key.split(AUTHZ_POLICY_ATTRIBUTES_SEPARATOR, 1)[0]
-        return cls._scope_registry.get(namespace, ScopeData)
+        return mcs.scope_registry.get(namespace, ScopeData)
 
-    def get_subclass_by_external_key(cls, external_key: str) -> Type["ScopeData"]:
+    @classmethod
+    def get_subclass_by_external_key(mcs, external_key: str) -> Type["ScopeData"]:
         """Get the appropriate subclass based on the format of external_key.
 
         Args:
@@ -145,12 +146,13 @@ def get_subclass_by_external_key(cls, external_key: str) -> Type["ScopeData"]:
         # 3. If the namespace is not recognized, we return the base ScopeData class
         # 4. The subclass implements a validation method to validate the entire key
         namespace = external_key.split(":", 1)[0]
-        scope_subclass = cls._scope_registry.get(namespace)
+        scope_subclass = mcs.scope_registry.get(namespace)
         if not scope_subclass or not scope_subclass.validate_external_key(external_key):
             return ScopeData  # Fallback to base class if not found or invalid
         return scope_subclass
 
-    def validate_external_key(cls, external_key: str) -> bool:
+    @classmethod
+    def validate_external_key(mcs, external_key: str) -> bool:
         """Validate the external_key format for the subclass.
 
         Args:
@@ -217,14 +219,14 @@ def validate_external_key(cls, external_key: str) -> bool:
 class SubjectMeta(type):
     """Metaclass for SubjectData to handle dynamic subclass instantiation based on namespace."""
 
-    _subject_registry: ClassVar[dict[str, Type["SubjectData"]]] = {}
+    subject_registry: ClassVar[dict[str, Type["SubjectData"]]] = {}
 
     def __init__(cls, name, bases, attrs):
         """Initialize the metaclass and register subclasses."""
         super().__init__(name, bases, attrs)
-        if not hasattr(cls, "_subject_registry"):
-            cls._subject_registry = {}
-        cls._subject_registry[cls.NAMESPACE] = cls
+        if not hasattr(cls, "subject_registry"):
+            cls.subject_registry = {}
+        cls.subject_registry[cls.NAMESPACE] = cls
 
     def __call__(cls, *args, **kwargs):
         """Instantiate the appropriate subclass based on the namespace in namespaced_key.
@@ -244,7 +246,8 @@ def __call__(cls, *args, **kwargs):
 
         return super().__call__(*args, **kwargs)
 
-    def get_subclass_by_namespaced_key(cls, namespaced_key: str) -> Type["SubjectData"]:
+    @classmethod
+    def get_subclass_by_namespaced_key(mcs, namespaced_key: str) -> Type["SubjectData"]:
         """Get the appropriate subclass based on the namespace in namespaced_key.
 
         Args:
@@ -254,7 +257,7 @@ def get_subclass_by_namespaced_key(cls, namespaced_key: str) -> Type["SubjectDat
             The subclass of SubjectData corresponding to the namespace, or SubjectData if not found.
         """
         namespace = namespaced_key.split(AUTHZ_POLICY_ATTRIBUTES_SEPARATOR, 1)[0]
-        return cls._subject_registry.get(namespace, SubjectData)
+        return mcs.subject_registry.get(namespace, SubjectData)
 
 
 @define

openedx_authz/api/permissions.py

@@ -5,13 +5,7 @@
 are not explicitly defined, but are inferred from the policy rules.
 """
 
-from openedx_authz.api.data import (
-    ActionData,
-    PermissionData,
-    PolicyIndex,
-    ScopeData,
-    SubjectData,
-)
+from openedx_authz.api.data import ActionData, PermissionData, PolicyIndex, ScopeData, SubjectData
 from openedx_authz.engine.enforcer import enforcer
 
 __all__ = [

openedx_authz/api/roles.py

@@ -16,10 +16,8 @@
     PolicyIndex,
     RoleAssignmentData,
     RoleData,
-    RoleMetadataData,
     ScopeData,
     SubjectData,
-    UserData,
 )
 from openedx_authz.api.permissions import get_permission_from_policy
 from openedx_authz.engine.enforcer import enforcer
@@ -163,9 +161,9 @@ def get_role_definitions_in_scope(scope: ScopeData) -> list[RoleData]:
     return [
         RoleData(
             namespaced_key=role,
-            permissions=permissions_per_role[role]["permissions"],
+            permissions=permissions["permissions"],
         )
-        for role in permissions_per_role.keys()
+        for role, permissions in permissions_per_role.items()
     ]
 
 
@@ -336,7 +334,7 @@ def get_subjects_role_assignments_for_role_in_scope(
             RoleAssignmentData(
                 subject=SubjectData(
                     namespaced_key=subject
-                ),  # TODO: I want this to behave like UserData or any other subclass of SubjectData depending on NAMESPACE
+                ),
                 role=RoleData(
                     external_key=role.external_key,
                     permissions=get_permissions_for_roles(role)[role.external_key][

openedx_authz/api/users.py

@@ -9,13 +9,7 @@
 (e.g., 'user@john_doe').
 """
 
-from openedx_authz.api.data import (
-    ActionData,
-    RoleAssignmentData,
-    RoleData,
-    ScopeData,
-    UserData,
-)
+from openedx_authz.api.data import ActionData, RoleAssignmentData, RoleData, ScopeData, UserData
 from openedx_authz.api.permissions import has_permission
 from openedx_authz.api.roles import (
     assign_role_to_subject_in_scope,

openedx_authz/engine/adapter.py

@@ -76,9 +76,7 @@ def is_filtered(self) -> bool:
         """
         return True
 
-    def load_filtered_policy(
-        self, model: Model, filter: Filter
-    ) -> None:  # pylint: disable=redefined-builtin
+    def load_filtered_policy(self, model: Model, filter: Filter) -> None:  # pylint: disable=redefined-builtin
         """
         Load policy rules from storage with filtering applied.
 
@@ -101,9 +99,7 @@ def load_filtered_policy(
         for line in filtered_queryset:
             persist.load_policy_line(str(line), model)
 
-    def filter_query(
-        self, queryset: QuerySet, filter: Filter
-    ) -> QuerySet:  # pylint: disable=redefined-builtin
+    def filter_query(self, queryset: QuerySet, filter: Filter) -> QuerySet:  # pylint: disable=redefined-builtin
         """
         Apply filter criteria to the policy queryset.
 

openedx_authz/engine/utils.py

@@ -5,7 +5,6 @@
 """
 
 import logging
-import os
 
 from casbin import Enforcer
 
@@ -25,7 +24,7 @@ def migrate_policy_from_file_to_db(
         target_enforcer (Enforcer): The Casbin enforcer instance to migrate policies to (database).
     """
     try:
-        # TODO: need to avoid loading twice the same policies
+        # Load latest policies from the source enforcer
         source_enforcer.load_policy()
         policies = source_enforcer.get_policy()
         for policy in policies:
@@ -34,7 +33,6 @@ def migrate_policy_from_file_to_db(
 
         for grouping_policy_ptype in GROUPING_POLICY_PTYPES:
             try:
-                source_enforcer.load_policy()
                 grouping_policies = source_enforcer.get_named_grouping_policy(
                     grouping_policy_ptype
                 )

openedx_authz/engine/watcher.py

@@ -48,7 +48,7 @@ def create_watcher():
         watcher = new_watcher(watcher_options)
         logger.info("Redis watcher created successfully")
         return watcher
-    except Exception as e:
+    except Exception as e:  # pylint: disable=broad-exception-caught
         logger.error(f"Failed to create Redis watcher: {e}")
         return None
 

openedx_authz/management/commands/load_policies.py

@@ -10,7 +10,7 @@
 """
 
 import casbin
-from django.core.management.base import BaseCommand, CommandError
+from django.core.management.base import BaseCommand
 
 from openedx_authz.engine.enforcer import enforcer as global_enforcer
 from openedx_authz.engine.utils import migrate_policy_from_file_to_db

openedx_authz/models.py

@@ -1,4 +1,5 @@
-"""Database models for the authorization framework.
+"""
+Database models for the authorization framework.
 
 These models will be used to store additional data about roles and permissions
 that are not natively supported by Casbin, so as to avoid modifying the Casbin

openedx_authz/settings/test.py

@@ -4,8 +4,6 @@
 
 import os
 
-from django.conf import settings
-
 from openedx_authz import ROOT_DIRECTORY
 
 # Add Casbin configuration