refactor: run ruff with compliant config

Author: mariajgrimaldi

Makefile

@@ -59,8 +59,8 @@ quality: ## check coding style with pycodestyle and pylint
 	tox -e quality
 
 format: ## format code with black and isort
-	black openedx_authz tests
-	isort openedx_authz tests
+	ruff format openedx_authz tests --line-length 120
+	ruff check --select I --fix --line-length 120
 
 pii_check: ## check for PII annotations on all Django models
 	tox -e pii_check

openedx_authz/api/data.py

@@ -248,9 +248,7 @@ def get_subclass_by_external_key(mcs, external_key: str) -> Type["ScopeData"]:
         scope_subclass = mcs.scope_registry.get(namespace)
 
         if not scope_subclass:
-            raise ValueError(
-                f"Unknown scope: {namespace} for external_key: {external_key}"
-            )
+            raise ValueError(f"Unknown scope: {namespace} for external_key: {external_key}")
 
         if not scope_subclass.validate_external_key(external_key):
             raise ValueError(f"Invalid external_key format: {external_key}")
@@ -281,9 +279,7 @@ def validate_external_key(mcs, external_key: str) -> bool:
         Returns:
             bool: True if valid, False otherwise.
         """
-        raise NotImplementedError(
-            "Subclasses must implement validate_external_key method."
-        )
+        raise NotImplementedError("Subclasses must implement validate_external_key method.")
 
 
 @define

openedx_authz/api/permissions.py

@@ -43,9 +43,7 @@ def get_all_permissions_in_scope(scope: ScopeData) -> list[PermissionData]:
         list of PermissionData: A list of PermissionData objects associated with the given scope.
     """
     enforcer = AuthzEnforcer.get_enforcer()
-    actions = enforcer.get_filtered_policy(
-        PolicyIndex.SCOPE.value, scope.namespaced_key
-    )
+    actions = enforcer.get_filtered_policy(PolicyIndex.SCOPE.value, scope.namespaced_key)
     return [get_permission_from_policy(action) for action in actions]
 
 
@@ -66,6 +64,4 @@ def is_subject_allowed(
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    return enforcer.enforce(
-        subject.namespaced_key, action.namespaced_key, scope.namespaced_key
-    )
+    return enforcer.enforce(subject.namespaced_key, action.namespaced_key, scope.namespaced_key)

openedx_authz/api/roles.py

@@ -78,9 +78,7 @@ def get_permissions_for_roles(
     permissions_by_role = {}
 
     for role in roles:
-        permissions_by_role[role.external_key] = {
-            "permissions": get_permissions_for_single_role(role)
-        }
+        permissions_by_role[role.external_key] = {"permissions": get_permissions_for_single_role(role)}
 
     return permissions_by_role
 
@@ -117,22 +115,15 @@ def get_permissions_for_active_roles_in_scope(
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    filtered_policy = enforcer.get_filtered_grouping_policy(
-        GroupingPolicyIndex.SCOPE.value, scope.namespaced_key
-    )
+    filtered_policy = enforcer.get_filtered_grouping_policy(GroupingPolicyIndex.SCOPE.value, scope.namespaced_key)
 
     if role:
         filtered_policy = [
-            policy
-            for policy in filtered_policy
-            if policy[GroupingPolicyIndex.ROLE.value] == role.namespaced_key
+            policy for policy in filtered_policy if policy[GroupingPolicyIndex.ROLE.value] == role.namespaced_key
         ]
 
     return get_permissions_for_roles(
-        [
-            RoleData(namespaced_key=policy[GroupingPolicyIndex.ROLE.value])
-            for policy in filtered_policy
-        ]
+        [RoleData(namespaced_key=policy[GroupingPolicyIndex.ROLE.value]) for policy in filtered_policy]
     )
 
 
@@ -150,9 +141,7 @@ def get_role_definitions_in_scope(scope: ScopeData) -> list[RoleData]:
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    policy_filtered = enforcer.get_filtered_policy(
-        PolicyIndex.SCOPE.value, scope.namespaced_key
-    )
+    policy_filtered = enforcer.get_filtered_policy(PolicyIndex.SCOPE.value, scope.namespaced_key)
 
     permissions_per_role = defaultdict(
         lambda: {
@@ -164,9 +153,7 @@ def get_role_definitions_in_scope(scope: ScopeData) -> list[RoleData]:
         permissions_per_role[policy[PolicyIndex.ROLE.value]]["scopes"].append(
             ScopeData(namespaced_key=policy[PolicyIndex.SCOPE.value])
         )  # TODO: I don't think this actually gets used anywhere
-        permissions_per_role[policy[PolicyIndex.ROLE.value]]["permissions"].append(
-            get_permission_from_policy(policy)
-        )
+        permissions_per_role[policy[PolicyIndex.ROLE.value]]["permissions"].append(get_permission_from_policy(policy))
 
     return [
         RoleData(
@@ -197,14 +184,10 @@ def get_all_roles_in_scope(scope: ScopeData) -> list[list[str]]:
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    return enforcer.get_filtered_grouping_policy(
-        GroupingPolicyIndex.SCOPE.value, scope.namespaced_key
-    )
+    return enforcer.get_filtered_grouping_policy(GroupingPolicyIndex.SCOPE.value, scope.namespaced_key)
 
 
-def assign_role_to_subject_in_scope(
-    subject: SubjectData, role: RoleData, scope: ScopeData
-) -> bool:
+def assign_role_to_subject_in_scope(subject: SubjectData, role: RoleData, scope: ScopeData) -> bool:
     """Assign a role to a subject.
 
     Args:
@@ -224,9 +207,7 @@ def assign_role_to_subject_in_scope(
     )
 
 
-def batch_assign_role_to_subjects_in_scope(
-    subjects: list[SubjectData], role: RoleData, scope: ScopeData
-) -> None:
+def batch_assign_role_to_subjects_in_scope(subjects: list[SubjectData], role: RoleData, scope: ScopeData) -> None:
     """Assign a role to a list of subjects.
 
     Args:
@@ -237,9 +218,7 @@ def batch_assign_role_to_subjects_in_scope(
         assign_role_to_subject_in_scope(subject, role, scope)
 
 
-def unassign_role_from_subject_in_scope(
-    subject: SubjectData, role: RoleData, scope: ScopeData
-) -> bool:
+def unassign_role_from_subject_in_scope(subject: SubjectData, role: RoleData, scope: ScopeData) -> bool:
     """Unassign a role from a subject.
 
     Args:
@@ -252,14 +231,10 @@ def unassign_role_from_subject_in_scope(
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    return enforcer.delete_roles_for_user_in_domain(
-        subject.namespaced_key, role.namespaced_key, scope.namespaced_key
-    )
+    return enforcer.delete_roles_for_user_in_domain(subject.namespaced_key, role.namespaced_key, scope.namespaced_key)
 
 
-def batch_unassign_role_from_subjects_in_scope(
-    subjects: list[SubjectData], role: RoleData, scope: ScopeData
-) -> None:
+def batch_unassign_role_from_subjects_in_scope(subjects: list[SubjectData], role: RoleData, scope: ScopeData) -> None:
     """Unassign a role from a list of subjects.
 
     Args:
@@ -282,9 +257,7 @@ def get_subject_role_assignments(subject: SubjectData) -> list[RoleAssignmentDat
     """
     enforcer = AuthzEnforcer.get_enforcer()
     role_assignments = []
-    for policy in enforcer.get_filtered_grouping_policy(
-        GroupingPolicyIndex.SUBJECT.value, subject.namespaced_key
-    ):
+    for policy in enforcer.get_filtered_grouping_policy(GroupingPolicyIndex.SUBJECT.value, subject.namespaced_key):
         role = RoleData(namespaced_key=policy[GroupingPolicyIndex.ROLE.value])
         role.permissions = get_permissions_for_single_role(role)
 
@@ -298,9 +271,7 @@ def get_subject_role_assignments(subject: SubjectData) -> list[RoleAssignmentDat
     return role_assignments
 
 
-def get_subject_role_assignments_in_scope(
-    subject: SubjectData, scope: ScopeData
-) -> list[RoleAssignmentData]:
+def get_subject_role_assignments_in_scope(subject: SubjectData, scope: ScopeData) -> list[RoleAssignmentData]:
     """Get the roles for a subject in a specific scope.
 
     Args:
@@ -314,9 +285,7 @@ def get_subject_role_assignments_in_scope(
     enforcer.load_policy()
     # TODO: we still need to get the remaining data for the role like email, etc
     role_assignments = []
-    for namespaced_key in enforcer.get_roles_for_user_in_domain(
-        subject.namespaced_key, scope.namespaced_key
-    ):
+    for namespaced_key in enforcer.get_roles_for_user_in_domain(subject.namespaced_key, scope.namespaced_key):
         role = RoleData(namespaced_key=namespaced_key)
         role_assignments.append(
             RoleAssignmentData(
@@ -333,9 +302,7 @@ def get_subject_role_assignments_in_scope(
     return role_assignments
 
 
-def get_subject_role_assignments_for_role_in_scope(
-    role: RoleData, scope: ScopeData
-) -> list[RoleAssignmentData]:
+def get_subject_role_assignments_for_role_in_scope(role: RoleData, scope: ScopeData) -> list[RoleAssignmentData]:
     """Get the subjects assigned to a specific role in a specific scope.
 
     Args:
@@ -347,9 +314,7 @@ def get_subject_role_assignments_for_role_in_scope(
     """
     enforcer = AuthzEnforcer.get_enforcer()
     role_assignments = []
-    for subject in enforcer.get_users_for_role_in_domain(
-        role.namespaced_key, scope.namespaced_key
-    ):
+    for subject in enforcer.get_users_for_role_in_domain(role.namespaced_key, scope.namespaced_key):
         if subject.startswith(f"{RoleData.NAMESPACE}{RoleData.SEPARATOR}"):
             # Skip roles that are also subjects
             continue
@@ -413,10 +378,5 @@ def get_subjects_for_role(role: RoleData) -> list[SubjectData]:
     """
     enforcer = AuthzEnforcer.get_enforcer()
     enforcer.load_policy()
-    policies = enforcer.get_filtered_grouping_policy(
-        GroupingPolicyIndex.ROLE.value, role.namespaced_key
-    )
-    return [
-        SubjectData(namespaced_key=policy[GroupingPolicyIndex.SUBJECT.value])
-        for policy in policies
-    ]
+    policies = enforcer.get_filtered_grouping_policy(GroupingPolicyIndex.ROLE.value, role.namespaced_key)
+    return [SubjectData(namespaced_key=policy[GroupingPolicyIndex.SUBJECT.value]) for policy in policies]

openedx_authz/api/users.py

@@ -37,9 +37,7 @@
 ]
 
 
-def assign_role_to_user_in_scope(
-    user_external_key: str, role_external_key: str, scope_external_key: str
-) -> bool:
+def assign_role_to_user_in_scope(user_external_key: str, role_external_key: str, scope_external_key: str) -> bool:
     """Assign a role to a user in a specific scope.
 
     Args:
@@ -57,9 +55,7 @@ def assign_role_to_user_in_scope(
     )
 
 
-def batch_assign_role_to_users_in_scope(
-    users: list[str], role_external_key: str, scope_external_key: str
-):
+def batch_assign_role_to_users_in_scope(users: list[str], role_external_key: str, scope_external_key: str):
     """Assign a role to multiple users in a specific scope.
 
     Args:
@@ -75,9 +71,7 @@ def batch_assign_role_to_users_in_scope(
     )
 
 
-def unassign_role_from_user(
-    user_external_key: str, role_external_key: str, scope_external_key: str
-):
+def unassign_role_from_user(user_external_key: str, role_external_key: str, scope_external_key: str):
     """Unassign a role from a user in a specific scope.
 
     Args:
@@ -95,9 +89,7 @@ def unassign_role_from_user(
     )
 
 
-def batch_unassign_role_from_users(
-    users: list[str], role_external_key: str, scope_external_key: str
-):
+def batch_unassign_role_from_users(users: list[str], role_external_key: str, scope_external_key: str):
     """Unassign a role from multiple users in a specific scope.
 
     Args:
@@ -125,9 +117,7 @@ def get_user_role_assignments(user_external_key: str) -> list[RoleAssignmentData
     return get_subject_role_assignments(UserData(external_key=user_external_key))
 
 
-def get_user_role_assignments_in_scope(
-    user_external_key: str, scope_external_key: str
-) -> list[RoleAssignmentData]:
+def get_user_role_assignments_in_scope(user_external_key: str, scope_external_key: str) -> list[RoleAssignmentData]:
     """Get the roles assigned to a user in a specific scope.
 
     Args:
@@ -172,9 +162,7 @@ def get_all_user_role_assignments_in_scope(
     Returns:
         list[RoleAssignmentData]: A list of user role assignments and all their metadata in the specified scope.
     """
-    return get_all_subject_role_assignments_in_scope(
-        ScopeData(external_key=scope_external_key)
-    )
+    return get_all_subject_role_assignments_in_scope(ScopeData(external_key=scope_external_key))
 
 
 def is_user_allowed(

openedx_authz/engine/adapter.py

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

openedx_authz/engine/enforcer.py

@@ -91,19 +91,15 @@ def _initialize_enforcer() -> FastEnforcer:
             # issues when the app is not fully loaded (e.g., while pulling translations, etc.).
             initialize_enforcer(db_alias)
         except Exception as e:
-            logger.error(
-                f"Failed to initialize Casbin enforcer with DB alias '{db_alias}': {e}"
-            )
+            logger.error(f"Failed to initialize Casbin enforcer with DB alias '{db_alias}': {e}")
             raise
 
         adapter = ExtendedAdapter()
         enforcer = FastEnforcer(settings.CASBIN_MODEL, adapter, enable_log=True)
         enforcer.enable_auto_save(True)
 
         if not Watcher:
-            logger.warning(
-                "Redis configuration not completed successfully. Watcher is disabled."
-            )
+            logger.warning("Redis configuration not completed successfully. Watcher is disabled.")
             return enforcer
 
         try:

openedx_authz/engine/utils.py

@@ -31,9 +31,7 @@ def migrate_policy_between_enforcers(
 
         # Load target enforcer policies to check for duplicates
         target_enforcer.load_policy()
-        logger.info(
-            f"Target enforcer has {len(target_enforcer.get_policy())} existing policies before migration."
-        )
+        logger.info(f"Target enforcer has {len(target_enforcer.get_policy())} existing policies before migration.")
 
         # TODO: this operations use the enforcer directly, which may not be ideal
         # since we have to load the policy after each addition to avoid duplicates.
@@ -53,31 +51,21 @@ def migrate_policy_between_enforcers(
 
         for grouping_policy_ptype in GROUPING_POLICY_PTYPES:
             try:
-                grouping_policies = source_enforcer.get_named_grouping_policy(
-                    grouping_policy_ptype
-                )
+                grouping_policies = source_enforcer.get_named_grouping_policy(grouping_policy_ptype)
                 for grouping in grouping_policies:
-                    if target_enforcer.has_named_grouping_policy(
-                        grouping_policy_ptype, *grouping
-                    ):
+                    if target_enforcer.has_named_grouping_policy(grouping_policy_ptype, *grouping):
                         logger.info(
                             f"Grouping policy {grouping_policy_ptype}, {grouping} already exists in target, skipping."
                         )
                         continue
-                    target_enforcer.add_named_grouping_policy(
-                        grouping_policy_ptype, *grouping
-                    )
+                    target_enforcer.add_named_grouping_policy(grouping_policy_ptype, *grouping)
 
                     # Ensure latest policies are loaded in the target enforcer after each addition
                     # to avoid duplicates
                     target_enforcer.load_policy()
             except KeyError as e:
-                logger.info(
-                    f"Skipping {grouping_policy_ptype} policies: {e} not found in source enforcer."
-                )
-        logger.info(
-            f"Successfully loaded policies from {source_enforcer.get_model()} into the database."
-        )
+                logger.info(f"Skipping {grouping_policy_ptype} policies: {e} not found in source enforcer.")
+        logger.info(f"Successfully loaded policies from {source_enforcer.get_model()} into the database.")
     except Exception as e:
         logger.error(f"Error loading policies from file: {e}")
         raise