feat: add batch_enforce (#46)

* feat: add v3-v5 field into enforce function's arg.

* to: fix ci

* to: fix ci

* to: fix ci lint check

* feat: add batch_enforce

* fix: linter

README.md

@@ -114,7 +114,8 @@ casdoor-python-sdk support basic user operations, like:
 - `get_users()`, get all users.
 - `modify_user(method: str, user: User)/add_user(user: User)/update_user(user: User)/delete_user(user: User)`, write user to database.
 - `refresh_token_request(refresh_token: str, scope: str)`, refresh access token
-- `enforce(self, permission_model_name: str, sub: str, obj: str, act: str)`, check permission from model
+- `enforce(self, permission_model_name: str, sub: str, obj: str, act: str, v3: Optional[str], v4: Optional[str], v5: Optional[str])`, check permission from model
+- `batch_enforce(self, permission_model_name: str, permission_rules: list[list[str]])`, batch check permission from model
 - `get_user_count(is_online: bool = None)`, get user count.
 
 ## Resource Owner Password Credentials Grant

src/casdoor/async_main.py

@@ -308,6 +308,59 @@ async def enforce(
 
             return has_permission
 
+    async def batch_enforce(
+            self,
+            permission_model_name: str,
+            permission_rules: list[list[str]]
+    ) -> list[bool]:
+        """
+        Send data to Casdoor enforce API
+
+        :param permission_model_name: Name permission model
+        :param permission_rules: permission rules to enforce
+                        [][0] -> sub: sub from Casbin
+                        [][1] -> obj: obj from Casbin
+                        [][2] -> act: act from Casbin
+                        [][3] -> v3: v3 from Casbin (optional)
+                        [][4] -> v4: v4 from Casbin (optional)
+                        [][5] -> v5: v5 from Casbin (optional)
+        """
+        url = self.endpoint + "/api/batch-enforce"
+        query_params = {
+            "clientId": self.client_id,
+            "clientSecret": self.client_secret
+        }
+
+        def map_rule(rule: list[str], idx) -> dict:
+            if len(rule) < 3:
+                raise ValueError("Invalid permission rule[{0}]: {1}"
+                                 .format(idx, rule))
+            result = {
+                "id": permission_model_name
+            }
+            for i in range(0, len(rule)):
+                result.update({"v{0}".format(i): rule[i]})
+            return result
+        params = [map_rule(permission_rules[i], i)
+                  for i in range(0, len(permission_rules))]
+        async with self._session.post(
+                url, params=query_params, json=params
+        ) as response:
+            if (
+                    response.status != 200 or
+                    "json" not in response.headers["content-type"]
+            ):
+                error_str = "Casdoor response error:\n" + str(response.text)
+                raise ValueError(error_str)
+
+            enforce_results = await response.json()
+
+            if not isinstance(enforce_results, bool):
+                error_str = "Casdoor response error:\n" + await response.text()
+                raise ValueError(error_str)
+
+            return enforce_results
+
     async def get_users(self) -> List[dict]:
         """
         Get the users from Casdoor.

src/casdoor/main.py

@@ -292,6 +292,56 @@ def enforce(
 
         return has_permission
 
+    def batch_enforce(
+            self,
+            permission_model_name: str,
+            permission_rules: list[list[str]]
+    ) -> list[bool]:
+        """
+        Send data to Casdoor enforce API
+
+        :param permission_model_name: Name permission model
+        :param permission_rules: permission rules to enforce
+                        [][0] -> sub: sub from Casbin
+                        [][1] -> obj: obj from Casbin
+                        [][2] -> act: act from Casbin
+                        [][3] -> v3: v3 from Casbin (optional)
+                        [][4] -> v4: v4 from Casbin (optional)
+                        [][5] -> v5: v5 from Casbin (optional)
+        """
+        url = self.endpoint + "/api/batch-enforce"
+        query_params = {
+            "clientId": self.client_id,
+            "clientSecret": self.client_secret
+        }
+
+        def map_rule(rule: list[str], idx) -> dict:
+            if len(rule) < 3:
+                raise ValueError("Invalid permission rule[{0}]: {1}"
+                                 .format(idx, rule))
+            result = {
+                "id": permission_model_name
+            }
+            for i in range(0, len(rule)):
+                result.update({"v{0}".format(i): rule[i]})
+            return result
+        params = [map_rule(permission_rules[i], i)
+                  for i in range(0, len(permission_rules))]
+        r = requests.post(url, json=params, params=query_params)
+        if r.status_code != 200 or "json" not in r.headers["content-type"]:
+            error_str = "Casdoor response error:\n" + str(r.text)
+            raise ValueError(error_str)
+
+        enforce_results = r.json()
+
+        if not isinstance(enforce_results, list) or \
+           len(enforce_results) == 0 or \
+           not isinstance(enforce_results[0], bool):
+            error_str = "Casdoor response error:\n" + r.text
+            raise ValueError(error_str)
+
+        return enforce_results
+
     def get_users(self) -> List[dict]:
         """
         Get the users from Casdoor.

src/tests/test_async_oauth.py

@@ -184,6 +184,56 @@ async def test_enforce_parmas(self, mock_post):
         )
         self.assertEqual(status, True)
 
+    def mocked_batch_enforce_requests_post(*args, **kwargs):
+        class MockResponse:
+            def __init__(self,
+                         json_data,
+                         status_code=200,
+                         headers={'content-type': 'json'}):
+                self.json_data = json_data
+                self.status_code = status_code
+                self.headers = headers
+
+            def json(self):
+                return self.json_data
+        json = kwargs.get('json')
+        result = [True for i in range(0, len(json))]
+        for k in range(0, len(json)):
+            for i in range(0, len(json[k]) - 1):
+                if json[k].get(f"v{i}") != f"v{i}":
+                    result[k] = False
+
+        return MockResponse(result)
+
+    @mock.patch("aiohttp.ClientSession.post",
+                side_effect=mocked_batch_enforce_requests_post)
+    def test_batch_enforce(self, mock_post):
+        sdk = self.get_sdk()
+        status = sdk.batch_enforce(
+            "built-in/permission-built-in",
+            [
+             ["v0", "v1", "v2", "v3", "v4", 'v5'],
+             ["v0", "v1", "v2", "v3", "v4", "v1"]
+            ]
+        )
+        self.assertEqual(len(status), 2)
+        self.assertEqual(status[0], True)
+        self.assertEqual(status[1], False)
+
+    @mock.patch("aiohttp.ClientSession.post",
+                side_effect=mocked_batch_enforce_requests_post)
+    def test_batch_enforce_raise(self, mock_post):
+        sdk = self.get_sdk()
+        with self.assertRaises(ValueError) as context:
+            sdk.batch_enforce(
+                "built-in/permission-built-in",
+                [
+                    ["v0", "v1"]
+                ]
+            )
+        self.assertEqual("Invalid permission rule[0]: ['v0', 'v1']",
+                         str(context.exception))
+
     async def test_get_users(self):
         sdk = self.get_sdk()
         users = await sdk.get_users()

src/tests/test_oauth.py

@@ -186,6 +186,56 @@ def test_enforce_parmas(self, mock_post):
         )
         self.assertEqual(status, True)
 
+    def mocked_batch_enforce_requests_post(*args, **kwargs):
+        class MockResponse:
+            def __init__(self,
+                         json_data,
+                         status_code=200,
+                         headers={'content-type': 'json'}):
+                self.json_data = json_data
+                self.status_code = status_code
+                self.headers = headers
+
+            def json(self):
+                return self.json_data
+        json = kwargs.get('json')
+        result = [True for i in range(0, len(json))]
+        for k in range(0, len(json)):
+            for i in range(0, len(json[k]) - 1):
+                if json[k].get(f"v{i}") != f"v{i}":
+                    result[k] = False
+
+        return MockResponse(result)
+
+    @mock.patch("requests.post",
+                side_effect=mocked_batch_enforce_requests_post)
+    def test_batch_enforce(self, mock_post):
+        sdk = self.get_sdk()
+        status = sdk.batch_enforce(
+            "built-in/permission-built-in",
+            [
+             ["v0", "v1", "v2", "v3", "v4", 'v5'],
+             ["v0", "v1", "v2", "v3", "v4", "v1"]
+            ]
+        )
+        self.assertEqual(len(status), 2)
+        self.assertEqual(status[0], True)
+        self.assertEqual(status[1], False)
+
+    @mock.patch("requests.post",
+                side_effect=mocked_batch_enforce_requests_post)
+    def test_batch_enforce_raise(self, mock_post):
+        sdk = self.get_sdk()
+        with self.assertRaises(ValueError) as context:
+            sdk.batch_enforce(
+                "built-in/permission-built-in",
+                [
+                    ["v0", "v1"]
+                ]
+            )
+        self.assertEqual("Invalid permission rule[0]: ['v0', 'v1']",
+                         str(context.exception))
+
     def test_get_users(self):
         sdk = self.get_sdk()
         users = sdk.get_users()