chore: implement format command for easier quality management

Author: mariajgrimaldi

Makefile

@@ -58,6 +58,10 @@ upgrade: ## update the requirements/*.txt files with the latest packages satisfy
 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
+
 pii_check: ## check for PII annotations on all Django models
 	tox -e pii_check
 

openedx_authz/api/roles.py

@@ -413,5 +413,10 @@ 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,7 +37,9 @@
 ]
 
 
-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:
@@ -55,7 +57,9 @@ def assign_role_to_user_in_scope(user_external_key: str, role_external_key: str,
     )
 
 
-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:
@@ -71,7 +75,9 @@ def batch_assign_role_to_users_in_scope(users: list[str], role_external_key: str
     )
 
 
-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:
@@ -89,7 +95,9 @@ def unassign_role_from_user(user_external_key: str, role_external_key: str, scop
     )
 
 
-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:
@@ -117,7 +125,9 @@ 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:
@@ -162,7 +172,9 @@ 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

@@ -76,7 +76,9 @@ 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  # pylint: disable=redefined-builtin
+    ) -> None:
         """
         Load policy rules from storage with filtering applied.
 
@@ -99,7 +101,9 @@ def load_filtered_policy(self, model: Model, filter: Filter) -> None:  # pylint:
         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  # pylint: disable=redefined-builtin
+    ) -> QuerySet:
         """
         Apply filter criteria to the policy queryset.
 

openedx_authz/engine/enforcer.py

@@ -91,15 +91,19 @@ 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,7 +31,9 @@ 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.

openedx_authz/management/commands/load_policies.py

@@ -69,7 +69,10 @@ def handle(self, *args, **options):
         Raises:
             CommandError: If the policy file is not found or loading fails.
         """
-        policy_file_path, model_file_path = options["policy_file_path"], options["model_file_path"]
+        policy_file_path, model_file_path = (
+            options["policy_file_path"],
+            options["model_file_path"],
+        )
         if policy_file_path is None:
             policy_file_path = os.path.join(
                 ROOT_DIRECTORY, "engine", "config", "authz.policy"
@@ -83,13 +86,25 @@ def handle(self, *args, **options):
 
         if options.get("clear_existing"):
             target_enforcer.load_policy()
-            if click.confirm(click.style('Do you want to delete existing roles? '
-                                         '(This will also delete the assignments related to those roles)',
-                                         fg='yellow', bold=True), default=False):
+            if click.confirm(
+                click.style(
+                    "Do you want to delete existing roles? "
+                    "(This will also delete the assignments related to those roles)",
+                    fg="yellow",
+                    bold=True,
+                ),
+                default=False,
+            ):
                 self._delete_existing_roles(target_enforcer)
 
-            if click.confirm(click.style('Do you want to delete existing permissions inheritance?',
-                                         fg='yellow', bold=True), default=False):
+            if click.confirm(
+                click.style(
+                    "Do you want to delete existing permissions inheritance?",
+                    fg="yellow",
+                    bold=True,
+                ),
+                default=False,
+            ):
                 self._delete_permissions_inheritance(target_enforcer)
 
         source_enforcer = casbin.Enforcer(model_file_path, policy_file_path)

openedx_authz/rest_api/decorators.py

@@ -39,7 +39,9 @@ def _decorator(func_or_class):
             SessionAuthenticationAllowInactiveUser,
         ]
         if is_authenticated:
-            func_or_class.permission_classes = [IsAuthenticated] + getattr(func_or_class, "permission_classes", [])
+            func_or_class.permission_classes = [IsAuthenticated] + getattr(
+                func_or_class, "permission_classes", []
+            )
         return func_or_class
 
     return _decorator

openedx_authz/rest_api/utils.py

@@ -28,7 +28,9 @@ def get_generic_scope(scope: ScopeData) -> ScopeData:
         >>> get_generic_scope(scope)
         ScopeData(namespaced_key="lib^*")
     """
-    return ScopeData(namespaced_key=f"{scope.NAMESPACE}{ScopeData.SEPARATOR}{GENERIC_SCOPE_WILDCARD}")
+    return ScopeData(
+        namespaced_key=f"{scope.NAMESPACE}{ScopeData.SEPARATOR}{GENERIC_SCOPE_WILDCARD}"
+    )
 
 
 def get_user_map(usernames: list[str]) -> dict[str, User]:
@@ -72,7 +74,9 @@ def get_user_by_username_or_email(username_or_email: str) -> User:
 
 
 def sort_users(
-    users: list[dict], sort_by: SortField = SortField.USERNAME, order: SortOrder = SortOrder.ASC
+    users: list[dict],
+    sort_by: SortField = SortField.USERNAME,
+    order: SortOrder = SortOrder.ASC,
 ) -> list[dict]:
     """
     Sort users by a given field and order.
@@ -90,16 +94,26 @@ def sort_users(
         list[dict]: The sorted users.
     """
     if sort_by not in SortField.values():
-        raise ValueError(f"Invalid field: '{sort_by}'. Must be one of {SortField.values()}")
+        raise ValueError(
+            f"Invalid field: '{sort_by}'. Must be one of {SortField.values()}"
+        )
 
     if order not in SortOrder.values():
-        raise ValueError(f"Invalid order: '{order}'. Must be one of {SortOrder.values()}")
-
-    sorted_users = sorted(users, key=lambda user: (user.get(sort_by) or "").lower(), reverse=order == SortOrder.DESC)
+        raise ValueError(
+            f"Invalid order: '{order}'. Must be one of {SortOrder.values()}"
+        )
+
+    sorted_users = sorted(
+        users,
+        key=lambda user: (user.get(sort_by) or "").lower(),
+        reverse=order == SortOrder.DESC,
+    )
     return sorted_users
 
 
-def filter_users(users: list[dict], search: str | None, roles: list[str] | None) -> list[dict]:
+def filter_users(
+    users: list[dict], search: str | None, roles: list[str] | None
+) -> list[dict]:
     """
     Filter users by a case-insensitive search string and/or by roles.
 
@@ -117,7 +131,10 @@ def filter_users(users: list[dict], search: str | None, roles: list[str] | None)
     filtered_users = []
     for user in users:
         if search:
-            matches_search = any(search in (user.get(field) or "").lower() for field in SearchField.values())
+            matches_search = any(
+                search in (user.get(field) or "").lower()
+                for field in SearchField.values()
+            )
             if not matches_search:
                 continue
 

openedx_authz/rest_api/v1/fields.py

@@ -8,7 +8,11 @@ class CommaSeparatedListField(serializers.CharField):
 
     def to_internal_value(self, data):
         """Convert string separated by commas to list of unique items preserving order"""
-        return list(dict.fromkeys(item.strip().lower() for item in data.split(",") if item.strip()))
+        return list(
+            dict.fromkeys(
+                item.strip().lower() for item in data.split(",") if item.strip()
+            )
+        )
 
     def to_representation(self, value):
         """Convert list to string separated by commas"""