feat: validate store_id and authorization_model_id are in a valid format

Author: rhamzeh

.openapi-generator/FILES

@@ -144,6 +144,8 @@ openfga_sdk/models/write_authorization_model_request.py
 openfga_sdk/models/write_authorization_model_response.py
 openfga_sdk/models/write_request.py
 openfga_sdk/rest.py
+openfga_sdk/test/validation.py
+openfga_sdk/validation.py
 requirements.txt
 setup.cfg
 setup.py

openfga_sdk/client/configuration.py

@@ -12,6 +12,8 @@
 """
 
 from openfga_sdk.configuration import Configuration
+from openfga_sdk.exceptions import FgaValidationException
+from openfga_sdk.validation import is_well_formed_ulid_string
 
 
 class ClientConfiguration(Configuration):
@@ -30,6 +32,13 @@ def __init__(
         super().__init__(api_scheme, api_host, store_id, credentials, retry_params)
         self._authorization_model_id = authorization_model_id
 
+    def is_valid(self):
+        super().is_valid()
+
+        if self.authorization_model_id is not None and self.authorization_model_id != "" and is_well_formed_ulid_string(self.authorization_model_id) is False:
+            raise FgaValidationException(
+                "authorization_model_id ('%s') is not in a valid ulid format" % self.authorization_model_id)
+
     @property
     def authorization_model_id(self):
         return self._authorization_model_id

openfga_sdk/client/openfga_client.py

@@ -24,6 +24,7 @@
 from openfga_sdk.client.write_transaction import WriteTransaction
 from openfga_sdk.client.configuration import ClientConfiguration
 from openfga_sdk.client.read_changes_body import ReadChangesBody
+from openfga_sdk.exceptions import FgaValidationException
 from openfga_sdk.models.assertion import Assertion
 from openfga_sdk.models.check_request import CheckRequest
 from openfga_sdk.models.contextual_tuple_keys import ContextualTupleKeys
@@ -35,6 +36,7 @@
 from openfga_sdk.models.write_assertions_request import WriteAssertionsRequest
 from openfga_sdk.models.write_authorization_model_request import WriteAuthorizationModelRequest
 from openfga_sdk.models.write_request import WriteRequest
+from openfga_sdk.validation import is_well_formed_ulid_string
 
 import asyncio
 import uuid
@@ -118,15 +120,20 @@ async def __aexit__(self, exc_type, exc_value, traceback):
     async def close(self):
         await self._api.close()
 
-    def _get_authorization_model_id(self, options):
+    def _get_authorization_model_id(self, options: object) -> str | None:
         """
         Return the authorization model ID if specified in the options.
         Otherwise return the authorization model ID stored in the client's configuration
         """
+        authorization_model_id = self._client_configuration.authorization_model_id
         if options is not None and "authorization_model_id" in options:
-            return options["authorization_model_id"]
-        else:
-            return self._client_configuration.authorization_model_id
+            authorization_model_id = options["authorization_model_id"]
+        if authorization_model_id is None or authorization_model_id == "":
+            return None
+        if is_well_formed_ulid_string(authorization_model_id) is False:
+            raise FgaValidationException(
+                "authorization_model_id ('%s') is not in a valid ulid format" % authorization_model_id)
+        return authorization_model_id
 
     def set_store_id(self, value):
         """

openfga_sdk/configuration.py

@@ -21,6 +21,7 @@
 from six.moves import http_client as httplib
 from urllib.parse import urlparse
 from openfga_sdk.exceptions import FgaValidationException, ApiValueError
+from openfga_sdk.validation import is_well_formed_ulid_string
 
 
 JSON_SCHEMA_VALIDATION_KEYWORDS = {
@@ -519,6 +520,11 @@ def is_valid(self):
         if (parsed_url.query != ''):
             raise ApiValueError(
                 'api_host `{}` is not expected to have query specified'.format(self.api_scheme))
+
+        if self.store_id is not None and self.store_id != "" and is_well_formed_ulid_string(self.store_id) is False:
+            raise FgaValidationException(
+                "store_id ('%s') is not in a valid ulid format" % self.store_id)
+
         if self._credentials is not None:
             self._credentials.validate_credentials_config()
 

openfga_sdk/validation.py

@@ -0,0 +1,25 @@
+# coding: utf-8
+
+"""
+   Python SDK for OpenFGA
+
+   API version: 0.1
+   Website: https://openfga.dev
+   Documentation: https://openfga.dev/docs
+   Support: https://discord.gg/8naAwJfWN6
+   License: [Apache-2.0](https://github.com/openfga/python-sdk/blob/main/LICENSE)
+
+   NOTE: This file was auto generated by OpenAPI Generator (https://openapi-generator.tech). DO NOT EDIT.
+"""
+
+import re
+
+
+def is_well_formed_ulid_string(ulid):
+    regex = re.compile("^[0-7][0-9A-HJKMNP-TV-Z]{25}$")
+    if not isinstance(ulid, str):
+        return False
+    match = regex.match(ulid)
+    if match is None:
+        return False
+    return True

test/test_open_fga_api.py

@@ -67,7 +67,7 @@
 from openfga_sdk.models.write_authorization_model_response import WriteAuthorizationModelResponse
 from openfga_sdk.models.write_request import WriteRequest
 
-store_id = 'd12345abc'
+store_id = '01H0H015178Y2V4CX10C2KGHF4'
 request_id = 'x1y2z3'
 
 # Helper function to construct mock response
@@ -134,7 +134,7 @@ async def test_check(self, mock_request):
             # Make sure the API was called with the right data
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/check',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/check',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -197,7 +197,7 @@ async def test_delete_store(self, mock_request):
             await api_instance.delete_store()
             mock_request.assert_called_once_with(
                 'DELETE',
-                'http://api.fga.example/stores/d12345abc',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4',
                 headers=ANY,
                 query_params=[],
                 body=None,
@@ -239,7 +239,7 @@ async def test_expand(self, mock_request):
             self.assertEqual(api_response, expected_response)
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/expand',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/expand',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -257,7 +257,7 @@ async def test_get_store(self, mock_request):
         Get a store  # noqa: E501
         """
         response_body = '''{
-  "id": "d12345abc",
+  "id": "01H0H015178Y2V4CX10C2KGHF4",
   "name": "test_store",
   "created_at": "2022-07-25T20:45:10.485Z",
   "updated_at": "2022-07-25T20:45:10.485Z"
@@ -271,11 +271,11 @@ async def test_get_store(self, mock_request):
             # Get a store
             api_response = await api_instance.get_store()
             self.assertIsInstance(api_response, GetStoreResponse)
-            self.assertEqual(api_response.id, 'd12345abc')
+            self.assertEqual(api_response.id, '01H0H015178Y2V4CX10C2KGHF4')
             self.assertEqual(api_response.name, 'test_store')
             mock_request.assert_called_once_with(
                 'GET',
-                'http://api.fga.example/stores/d12345abc',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4',
                 headers=ANY,
                 query_params=[],
                 _preload_content=ANY,
@@ -313,7 +313,7 @@ async def test_list_objects(self, mock_request):
             self.assertEqual(api_response.objects, ['document:abcd1234'])
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/list-objects',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/list-objects',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -436,7 +436,7 @@ async def test_read(self, mock_request):
             self.assertEqual(api_response, expected_data)
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/read',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/read',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -488,7 +488,7 @@ async def test_read_assertions(self, mock_request):
             self.assertEqual(api_response.assertions, [assertion])
             mock_request.assert_called_once_with(
                 'GET',
-                'http://api.fga.example/stores/d12345abc/assertions/01G5JAVJ41T49E9TT3SKVS7X1J',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/assertions/01G5JAVJ41T49E9TT3SKVS7X1J',
                 headers=ANY,
                 query_params=[],
                 _preload_content=ANY,
@@ -573,7 +573,7 @@ async def test_read_authorization_model(self, mock_request):
             self.assertEqual(api_response.authorization_model, authorization_model)
             mock_request.assert_called_once_with(
                 'GET',
-                'http://api.fga.example/stores/d12345abc/authorization-models/01G5JAVJ41T49E9TT3SKVS7X1J',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/authorization-models/01G5JAVJ41T49E9TT3SKVS7X1J',
                 headers=ANY,
                 query_params=[],
                 _preload_content=ANY,
@@ -628,7 +628,7 @@ async def test_read_changes(self, mock_request):
             self.assertEqual(api_response, read_changes)
             mock_request.assert_called_once_with(
                 'GET',
-                'http://api.fga.example/stores/d12345abc/changes',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/changes',
                 headers=ANY,
                 query_params=[('type', 'document'), ('page_size', 1),
                               ('continuation_token', 'abcdefg')],
@@ -670,7 +670,7 @@ async def test_write(self, mock_request):
             )
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/write',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/write',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -714,7 +714,7 @@ async def test_write_delete(self, mock_request):
             )
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/write',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/write',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -759,7 +759,7 @@ async def test_write_assertions(self, mock_request):
             )
             mock_request.assert_called_once_with(
                 'PUT',
-                'http://api.fga.example/stores/d12345abc/assertions/xyz0123',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/assertions/xyz0123',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -819,7 +819,7 @@ async def test_write_authorization_model(self, mock_request):
             self.assertEqual(api_response, expected_response)
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/authorization-models',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/authorization-models',
                 headers=ANY,
                 query_params=[],
                 post_params=[],
@@ -906,6 +906,17 @@ def test_configuration_has_query(self):
         )
         self.assertRaises(ApiValueError, configuration.is_valid)
 
+    def test_configuration_store_id_invalid(self):
+        """
+        Test whether ApiValueError is raised if host has query
+        """
+        configuration = openfga_sdk.Configuration(
+            api_host='localhost',
+            api_scheme='http',
+            store_id="abcd"
+        )
+        self.assertRaises(FgaValidationException, configuration.is_valid)
+
     async def test_bad_configuration_read_authorization_model(self):
         """
         Test whether FgaValidationException is raised for API (reading authorization models)
@@ -1166,7 +1177,7 @@ async def test_check_api_token(self, mock_request):
                 {'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'openfga-sdk python/0.1.1', 'Authorization': 'Bearer TOKEN1'})
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/check',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/check',
                 headers=expectedHeader,
                 query_params=[],
                 post_params=[],
@@ -1209,7 +1220,7 @@ async def test_check_custom_header(self, mock_request):
                 {'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'openfga-sdk python/0.1.1', 'Custom Header': 'custom value'})
             mock_request.assert_called_once_with(
                 'POST',
-                'http://api.fga.example/stores/d12345abc/check',
+                'http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/check',
                 headers=expectedHeader,
                 query_params=[],
                 post_params=[],

test/test_openfga_client.py

@@ -30,7 +30,7 @@
 from openfga_sdk.client.read_changes_body import ReadChangesBody
 from openfga_sdk.client.single_write_response import SingleWriteResponse
 from openfga_sdk.client.write_transaction import WriteTransaction
-from openfga_sdk.exceptions import ValidationException
+from openfga_sdk.exceptions import ValidationException, FgaValidationException
 from openfga_sdk.models.assertion import Assertion
 from openfga_sdk.models.authorization_model import AuthorizationModel
 from openfga_sdk.models.check_response import CheckResponse
@@ -2006,3 +2006,26 @@ async def test_update_auth_model(self, mock_request):
                 _preload_content=ANY,
                 _request_timeout=None
             )
+
+    def test_configuration_store_id_invalid(self):
+        """
+        Test whether ApiValueError is raised if host has query
+        """
+        configuration = ClientConfiguration(
+            api_host='localhost',
+            api_scheme='http',
+            store_id="abcd"
+        )
+        self.assertRaises(FgaValidationException, configuration.is_valid)
+
+    def test_configuration_authorization_model_id_invalid(self):
+        """
+        Test whether ApiValueError is raised if host has query
+        """
+        configuration = ClientConfiguration(
+            api_host='localhost',
+            api_scheme='http',
+            store_id="01H15K9J85050XTEDPVM8DJM78",
+            authorization_model_id="abcd"
+        )
+        self.assertRaises(FgaValidationException, configuration.is_valid)

test/test_validation.py

@@ -0,0 +1,27 @@
+import unittest
+
+from openfga_sdk.validation import is_well_formed_ulid_string
+
+
+class TestValidation(unittest.TestCase):
+    """Test for validation"""
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_is_well_formed_ulid_string_valid_ulids(self):
+        self.assertEqual(is_well_formed_ulid_string("01H0GVCS1HCQM6SJRJ4A026FZ9"), True, "Should be True")
+        self.assertEqual(is_well_formed_ulid_string("01H0GVD9ACPFKGMWJV0Y93ZM7H"), True, "Should be True")
+        self.assertEqual(is_well_formed_ulid_string("01H0GVDH0FRZ4WAFED6T9KZYZR"), True, "Should be True")
+        self.assertEqual(is_well_formed_ulid_string("01H0GVDSW72AZ8QV3R0HJ91QBX"), True, "Should be True")
+
+    def test_is_well_formed_ulid_string_invalid_ulids(self):
+        self.assertEqual(is_well_formed_ulid_string("abc"), False, "Should be False")
+        self.assertEqual(is_well_formed_ulid_string(123), False, "Should be False")
+        self.assertEqual(is_well_formed_ulid_string(None), False, "Should be False")
+        self.assertEqual(is_well_formed_ulid_string("01H0GVDSW72AZ8QV3R0HJ91QBXa"), False, "Should be False")
+        self.assertEqual(is_well_formed_ulid_string("b523ad13-8adb-4803-a6db-013ac50197ca"), False, "Should be False")
+        self.assertEqual(is_well_formed_ulid_string("9240BFC0-DA00-457B-A328-FC370A598D60"), False, "Should be False")