feat: add support for using FGA Cache via fga_cache_url configuration (#656)

Author: yosiharan

README.md

@@ -1224,6 +1224,26 @@ relations = descope_client.mgmt.fga.check(
 )
 ```
 
+Response times of repeated FGA `check` calls, especially in high volume scenarios, can be reduced to sub-millisecond scales by re-directing the calls to a Descope FGA Cache Proxy running in the same backend cluster as your application.
+After setting up the proxy server via the Descope provided Docker image, set the `fga_cache_url` parameter to be equal to the proxy URL to enable its use in the SDK, as shown in the example below:
+
+```python
+# Initialize client with FGA cache URL
+descope_client = DescopeClient(
+    project_id="<Project ID>",
+    management_key="<Management Key>",
+    fga_cache_url="https://10.0.0.4",  # example FGA Cache Proxy URL, running inside the same backend cluster
+)
+```
+
+When the `fga_cache_url` is configured, the following FGA methods will automatically use the cache proxy instead of the default Descope API:
+- `save_schema`
+- `create_relations`  
+- `delete_relations`
+- `check`
+
+Other FGA operations like `load_schema` will continue to use the standard Descope API endpoints.
+
 ### Manage Project
 
 You can change the project name, as well as clone the current project to

descope/auth.py

@@ -74,6 +74,7 @@ def __init__(
         timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
         jwt_validation_leeway: int = 5,
         auth_management_key: str | None = None,
+        fga_cache_url: str | None = None,
     ):
         self.lock_public_keys = Lock()
         # validate project id
@@ -99,6 +100,7 @@ def __init__(
         self.auth_management_key = auth_management_key or os.getenv(
             "DESCOPE_AUTH_MANAGEMENT_KEY"
         )
+        self.fga_cache_url = fga_cache_url
 
         public_key = public_key or os.getenv("DESCOPE_PUBLIC_KEY")
         with self.lock_public_keys:
@@ -208,6 +210,31 @@ def do_delete(
         self._raise_from_response(response)
         return response
 
+    def do_post_with_custom_base_url(
+        self,
+        uri: str,
+        body: dict | list[dict] | list[str] | None,
+        custom_base_url: str | None = None,
+        params=None,
+        pswd: str | None = None,
+    ) -> requests.Response:
+        """
+        Post request with optional custom base URL.
+        If base_url is provided, use it instead of self.base_url.
+        """
+        effective_base_url = custom_base_url if custom_base_url else self.base_url
+        response = requests.post(
+            f"{effective_base_url}{uri}",
+            headers=self._get_default_headers(pswd),
+            json=body,
+            allow_redirects=False,
+            verify=self.secure,
+            params=params,
+            timeout=self.timeout_seconds,
+        )
+        self._raise_from_response(response)
+        return response
+
     def exchange_token(
         self, uri, code: str, audience: str | None | Iterable[str] = None
     ) -> dict:
@@ -637,13 +664,13 @@ def _validate_token(
                 audience=audience,
                 leeway=self.jwt_validation_leeway,
             )
-        except (ImmatureSignatureError):
+        except ImmatureSignatureError:
             raise AuthException(
                 400,
                 ERROR_TYPE_INVALID_TOKEN,
                 "Received Invalid token (nbf in future) during jwt validation. Error can be due to time glitch (between machines), try to set the jwt_validation_leeway parameter (in DescopeClient) to higher value than 5sec which is the default",
             )
-        except (ExpiredSignatureError):
+        except ExpiredSignatureError:
             raise AuthException(
                 401,
                 ERROR_TYPE_INVALID_TOKEN,

descope/descope_client.py

@@ -31,6 +31,7 @@ def __init__(
         timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
         jwt_validation_leeway: int = 5,
         auth_management_key: str | None = None,
+        fga_cache_url: str | None = None,
     ):
         auth = Auth(
             project_id,
@@ -40,6 +41,7 @@ def __init__(
             timeout_seconds,
             jwt_validation_leeway,
             auth_management_key,
+            fga_cache_url,
         )
         self._auth = auth
         self._mgmt = MGMT(auth)

descope/management/fga.py

@@ -40,9 +40,10 @@ def save_schema(self, schema: str):
         Raise:
         AuthException: raised if saving fails
         """
-        self._auth.do_post(
+        self._auth.do_post_with_custom_base_url(
             MgmtV1.fga_save_schema,
             {"dsl": schema},
+            custom_base_url=self._auth.fga_cache_url,
             pswd=self._auth.management_key,
         )
 
@@ -64,11 +65,12 @@ def create_relations(
         Raise:
         AuthException: raised if create relations fails
         """
-        self._auth.do_post(
+        self._auth.do_post_with_custom_base_url(
             MgmtV1.fga_create_relations,
             {
                 "tuples": relations,
             },
+            custom_base_url=self._auth.fga_cache_url,
             pswd=self._auth.management_key,
         )
 
@@ -83,11 +85,12 @@ def delete_relations(
         Raise:
         AuthException: raised if delete relations fails
         """
-        self._auth.do_post(
+        self._auth.do_post_with_custom_base_url(
             MgmtV1.fga_delete_relations,
             {
                 "tuples": relations,
             },
+            custom_base_url=self._auth.fga_cache_url,
             pswd=self._auth.management_key,
         )
 
@@ -124,11 +127,12 @@ def check(
         Raise:
         AuthException: raised if query fails
         """
-        response = self._auth.do_post(
+        response = self._auth.do_post_with_custom_base_url(
             MgmtV1.fga_check,
             {
                 "tuples": relations,
             },
+            custom_base_url=self._auth.fga_cache_url,
             pswd=self._auth.management_key,
         )
         return list(

tests/management/test_fga.py

@@ -308,3 +308,184 @@ def test_save_resources_details_error(self):
                 client.mgmt.fga.save_resources_details,
                 details,
             )
+
+    def test_fga_cache_url_save_schema(self):
+        # Test FGA cache URL functionality for save_schema
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            client.mgmt.fga.save_schema("model AuthZ 1.0")
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.fga_save_schema}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"dsl": "model AuthZ 1.0"},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_fga_cache_url_create_relations(self):
+        # Test FGA cache URL functionality for create_relations
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        relations = [
+            {
+                "resource": "r",
+                "resourceType": "rt",
+                "relation": "rel",
+                "target": "u",
+                "targetType": "ty",
+            }
+        ]
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            client.mgmt.fga.create_relations(relations)
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.fga_create_relations}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"tuples": relations},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_fga_cache_url_delete_relations(self):
+        # Test FGA cache URL functionality for delete_relations
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        relations = [
+            {
+                "resource": "r",
+                "resourceType": "rt",
+                "relation": "rel",
+                "target": "u",
+                "targetType": "ty",
+            }
+        ]
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            client.mgmt.fga.delete_relations(relations)
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.fga_delete_relations}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"tuples": relations},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_fga_cache_url_check(self):
+        # Test FGA cache URL functionality for check
+        fga_cache_url = "https://my-fga-cache.example.com"
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            fga_cache_url=fga_cache_url,
+        )
+
+        relations = [
+            {
+                "resource": "r",
+                "resourceType": "rt",
+                "relation": "rel",
+                "target": "u",
+                "targetType": "ty",
+            }
+        ]
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            mock_post.return_value.json.return_value = {
+                "tuples": [
+                    {
+                        "allowed": True,
+                        "tuple": relations[0],
+                    }
+                ]
+            }
+            result = client.mgmt.fga.check(relations)
+            mock_post.assert_called_with(
+                f"{fga_cache_url}{MgmtV1.fga_check}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"tuples": relations},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+            self.assertEqual(len(result), 1)
+            self.assertTrue(result[0]["allowed"])
+            self.assertEqual(result[0]["relation"], relations[0])
+
+    def test_fga_without_cache_url_uses_default_base_url(self):
+        # Test that FGA methods use default base URL when cache URL is not provided
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+            # No fga_cache_url provided
+        )
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            client.mgmt.fga.save_schema("model AuthZ 1.0")
+            # Should use default base URL
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.fga_save_schema}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"dsl": "model AuthZ 1.0"},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )