[FC-0099] refactor: disable enforcer auto save when content libraries V2 is off

CHANGELOG.rst

@@ -16,6 +16,14 @@ Unreleased
 
 *
 
+0.11.0 - 2025-10-29
+********************
+
+Added
+=====
+
+* Disable auto-save and auto-load of policies if Content Library V2 is disabled.
+
 0.10.1 - 2025-10-28
 ********************
 

openedx_authz/__init__.py

@@ -4,6 +4,6 @@
 
 import os
 
-__version__ = "0.10.1"
+__version__ = "0.11.0"
 
 ROOT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))

openedx_authz/engine/enforcer.py

@@ -23,6 +23,18 @@
 
 from openedx_authz.engine.adapter import ExtendedAdapter
 
+try:
+    from cms.djangoapps.contentstore.toggles import libraries_v2_enabled
+except ImportError:
+    # If the CMS is not available, define a dummy toggle that is always enabled
+    class DummyToggle:
+        @staticmethod
+        def is_enabled():
+            return True
+
+    libraries_v2_enabled = DummyToggle()
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -57,6 +69,84 @@ def __new__(cls):
             cls._enforcer = cls._initialize_enforcer()
         return cls._enforcer
 
+    @classmethod
+    def configure_enforcer_auto_loading(cls, auto_load_policy_interval: int = None):
+        """Enable auto-load policy and auto-save on the enforcer.
+
+        This method ensures that the singleton enforcer instance is created
+        and ready for use.
+
+        Returns:
+            None
+        """
+        if not cls._enforcer.is_auto_loading_running():
+            cls._enforcer.start_auto_load_policy(auto_load_policy_interval)
+
+    @classmethod
+    def is_auto_save_enabled(cls) -> bool:
+        """Check if auto-save is currently enabled on the enforcer.
+
+        Returns:
+            bool: True if auto-save is enabled, False otherwise
+        """
+        if cls._enforcer is None:
+            return False
+        return cls._enforcer._e.auto_save  # pylint: disable=protected-access
+
+    @classmethod
+    def configure_enforcer_auto_save(cls, auto_save_policy: bool):
+        """Configure auto-save on the enforcer.
+
+        This method ensures that auto-save is enabled or disabled based on
+        the auto_save_policy parameter.
+
+        Args:
+            auto_save_policy: True to enable auto-save, False to disable
+
+        Returns:
+            None
+        """
+        if cls.is_auto_save_enabled() != auto_save_policy:
+            cls._enforcer.enable_auto_save(auto_save_policy)
+
+    @classmethod
+    def deactivate_enforcer(cls):
+        """Deactivate the current enforcer instance, if any.
+
+        This method stops the auto-load policy thread. It can be used in testing
+        or when re-initialization of the enforcer is needed. IT DOES NOT
+        clear the singleton instance to avoid initializing it again unintentionally.
+
+        Returns:
+            None
+        """
+        if cls._enforcer is not None:
+            try:
+                cls._enforcer.stop_auto_load_policy()
+                cls._enforcer.enable_auto_save(False)
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                logger.error(f"Error stopping auto-load policy thread: {e}")
+
+    @classmethod
+    def configure_enforcer_auto_save_and_load(cls):
+        """Enable auto-load policy and auto-save on the enforcer.
+
+        This method ensures that the singleton enforcer instance is configured
+        for auto-load and auto-save based on settings.
+
+        Returns:
+            None
+        """
+        auto_load_policy_interval = getattr(settings, "CASBIN_AUTO_LOAD_POLICY_INTERVAL", 0)
+        auto_save_policy = getattr(settings, "CASBIN_AUTO_SAVE_POLICY", True)
+
+        if auto_load_policy_interval > 0:
+            cls.configure_enforcer_auto_loading(auto_load_policy_interval)
+        else:
+            logger.warning("CASBIN_AUTO_LOAD_POLICY_INTERVAL is not set or zero; auto-load is disabled.")
+
+        cls.configure_enforcer_auto_save(auto_save_policy)
+
     @classmethod
     def get_enforcer(cls) -> SyncedEnforcer:
         """Get the enforcer instance, creating it if needed.
@@ -66,10 +156,21 @@ def get_enforcer(cls) -> SyncedEnforcer:
         """
         if cls._enforcer is None:
             cls._enforcer = cls._initialize_enforcer()
+
+        # HACK: This code block will only be useful when in Ulmo to deactivate
+        # the enforcer when the new library experience is disabled. It should be
+        # removed for the next release cycle.
+        # When replaced, we will only need to configure the enforcer here. Which
+        # is in charge of enabling/disabling auto-load and auto-save.
+        if libraries_v2_enabled.is_enabled():
+            cls.configure_enforcer_auto_save_and_load()
+        else:
+            cls.deactivate_enforcer()
+
         return cls._enforcer
 
-    @staticmethod
-    def _initialize_enforcer() -> SyncedEnforcer:
+    @classmethod
+    def _initialize_enforcer(cls) -> SyncedEnforcer:
         """
         Create and configure the Casbin SyncedEnforcer instance.
 
@@ -93,12 +194,5 @@ def _initialize_enforcer() -> SyncedEnforcer:
 
         adapter = ExtendedAdapter()
         enforcer = SyncedEnforcer(settings.CASBIN_MODEL, adapter)
-        auto_load_policy_interval = getattr(settings, "CASBIN_AUTO_LOAD_POLICY_INTERVAL", 0)
-        if auto_load_policy_interval > 0:
-            enforcer.start_auto_load_policy(auto_load_policy_interval)
-            enforcer.enable_auto_save(True)
-        else:
-            # Disable auto-save to prevent unnecessary database writes
-            enforcer.enable_auto_save(False)
 
         return enforcer

openedx_authz/settings/common.py

@@ -20,7 +20,22 @@ def plugin_settings(settings):
     casbin_adapter_app = "openedx_authz.engine.apps.CasbinAdapterConfig"
     if casbin_adapter_app not in settings.INSTALLED_APPS:
         settings.INSTALLED_APPS.append(casbin_adapter_app)
-    # Add Casbin configuration
-    settings.CASBIN_MODEL = os.path.join(ROOT_DIRECTORY, "engine", "config", "model.conf")
+
+    # Casbin settings for model and policy synchronization
+
+    # Set default CASBIN_MODEL if not already set, this points to the model.conf file
+    # which defines the access control model for Casbin.
+    if not hasattr(settings, "CASBIN_MODEL"):
+        settings.CASBIN_MODEL = os.path.join(ROOT_DIRECTORY, "engine", "config", "model.conf")
+
+    # Set default CASBIN_AUTO_LOAD_POLICY_INTERVAL if not already set.
+    # This setting defines how often (in seconds) the Casbin enforcer should
+    # automatically reload policies from the database.
     if not hasattr(settings, "CASBIN_AUTO_LOAD_POLICY_INTERVAL"):
         settings.CASBIN_AUTO_LOAD_POLICY_INTERVAL = 5
+
+    # Set default CASBIN_AUTO_SAVE_POLICY if not already set.
+    # This setting defines whether the Casbin enforcer should automatically
+    # save policy changes back to the database.
+    if not hasattr(settings, "CASBIN_AUTO_SAVE_POLICY"):
+        settings.CASBIN_AUTO_SAVE_POLICY = True

openedx_authz/settings/test.py

@@ -70,3 +70,4 @@ def plugin_settings(settings):  # pylint: disable=unused-argument
 # Casbin configuration
 CASBIN_MODEL = os.path.join(ROOT_DIRECTORY, "engine", "config", "model.conf")
 CASBIN_AUTO_LOAD_POLICY_INTERVAL = 0
+CASBIN_AUTO_SAVE_POLICY = True