fix: custom header precedence

README.md

@@ -6,7 +6,7 @@
 [![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2Fopenfga%2Fpython-sdk.svg?type=shield)](https://app.fossa.com/projects/git%2Bgithub.com%2Fopenfga%2Fpython-sdk?ref=badge_shield)
 [![OpenSSF Scorecard](https://api.securityscorecards.dev/projects/github.com/openfga/python-sdk/badge)](https://securityscorecards.dev/viewer/?uri=github.com/openfga/python-sdk)
 [![Join our community](https://img.shields.io/badge/slack-cncf_%23openfga-40abb8.svg?logo=slack)](https://openfga.dev/community)
-[![X](https://img.shields.io/twitter/follow/openfga?color=%23179CF0&logo=twitter&style=flat-square "@openfga on Twitter")](https://x.com/openfga)
+[![X](https://img.shields.io/twitter/follow/openfga?color=%23179CF0&logo=x&style=flat-square "@openfga on X")](https://x.com/openfga)
 
 This is an autogenerated python SDK for OpenFGA. It provides a wrapper around the [OpenFGA API definition](https://openfga.dev/api).
 
@@ -64,7 +64,7 @@ OpenFGA is designed to make it easy for application builders to model their perm
 
 - [OpenFGA Documentation](https://openfga.dev/docs)
 - [OpenFGA API Documentation](https://openfga.dev/api/service)
-- [Twitter](https://twitter.com/openfga)
+- [X](https://x.com/openfga)
 - [OpenFGA Community](https://openfga.dev/community)
 - [Zanzibar Academy](https://zanzibar.academy)
 - [Google's Zanzibar Paper (2019)](https://research.google/pubs/pub48190/)

openfga_sdk/api_client.py

@@ -178,8 +178,7 @@ async def __call_api(
         start = float(time.time())
 
         # header parameters
-        header_params = header_params or {}
-        header_params.update(self.default_headers)
+        header_params = {**self.default_headers, **(header_params or {})}
         if self.cookie:
             header_params["Cookie"] = self.cookie
         if header_params:

openfga_sdk/client/client.py

@@ -106,7 +106,7 @@ def set_heading_if_not_set(
         _options["headers"] = {}
 
     if type(_options["headers"]) is dict:
-        if type(_options["headers"].get(name)) not in [int, str]:
+        if _options["headers"].get(name) is None:
             _options["headers"][name] = value
 
     return _options

openfga_sdk/sync/api_client.py

@@ -177,8 +177,7 @@ def __call_api(
         start = float(time.time())
 
         # header parameters
-        header_params = header_params or {}
-        header_params.update(self.default_headers)
+        header_params = {**self.default_headers, **(header_params or {})}
         if self.cookie:
             header_params["Cookie"] = self.cookie
         if header_params:

openfga_sdk/sync/client/client.py

@@ -107,7 +107,7 @@ def set_heading_if_not_set(
         _options["headers"] = {}
 
     if type(_options["headers"]) is dict:
-        if type(_options["headers"].get(name)) not in [int, str]:
+        if _options["headers"].get(name) is None:
             _options["headers"][name] = value
 
     return _options

test/api/open_fga_api_test.py

@@ -2048,6 +2048,123 @@ async def test_read_with_type_only_object(self, mock_request):
             )
             await api_client.close()
 
+    @patch.object(rest.RESTClientObject, "request")
+    async def test_check_custom_header_override_default_header(self, mock_request):
+        """Test case for per-request custom header overriding default header
+
+        Per-request custom headers should override default headers with the same name
+        """
+
+        # First, mock the response
+        response_body = '{"allowed": true}'
+        mock_request.return_value = mock_response(response_body, 200)
+
+        configuration = self.configuration
+        configuration.store_id = store_id
+        async with openfga_sdk.ApiClient(configuration) as api_client:
+            # Set a default header
+            api_client.set_default_header("X-Custom-Header", "default-value")
+            api_instance = open_fga_api.OpenFgaApi(api_client)
+            body = CheckRequest(
+                tuple_key=TupleKey(
+                    object="document:2021-budget",
+                    relation="reader",
+                    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
+                ),
+            )
+            # Make request with per-request custom header that should override the default
+            api_response = await api_instance.check(
+                body=body,
+                _headers={"X-Custom-Header": "per-request-value"},
+            )
+            self.assertIsInstance(api_response, CheckResponse)
+            self.assertTrue(api_response.allowed)
+            # Make sure the API was called with the per-request header value, not the default
+            expected_headers = urllib3.response.HTTPHeaderDict(
+                {
+                    "Accept": "application/json",
+                    "Content-Type": "application/json",
+                    "User-Agent": "openfga-sdk python/0.9.6",
+                    "X-Custom-Header": "per-request-value",  # Should be the per-request value
+                }
+            )
+            mock_request.assert_called_once_with(
+                "POST",
+                "http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/check",
+                headers=expected_headers,
+                query_params=[],
+                post_params=[],
+                body={
+                    "tuple_key": {
+                        "object": "document:2021-budget",
+                        "relation": "reader",
+                        "user": "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
+                    }
+                },
+                _preload_content=ANY,
+                _request_timeout=None,
+            )
+
+    @patch.object(rest.RESTClientObject, "request")
+    async def test_check_per_request_header_and_default_header_coexist(
+        self, mock_request
+    ):
+        """Test case for per-request custom header and default header coexisting
+
+        Per-request custom headers should be merged with default headers
+        """
+
+        # First, mock the response
+        response_body = '{"allowed": true}'
+        mock_request.return_value = mock_response(response_body, 200)
+
+        configuration = self.configuration
+        configuration.store_id = store_id
+        async with openfga_sdk.ApiClient(configuration) as api_client:
+            # Set a default header
+            api_client.set_default_header("X-Default-Header", "default-value")
+            api_instance = open_fga_api.OpenFgaApi(api_client)
+            body = CheckRequest(
+                tuple_key=TupleKey(
+                    object="document:2021-budget",
+                    relation="reader",
+                    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
+                ),
+            )
+            # Make request with per-request custom header (different from default)
+            api_response = await api_instance.check(
+                body=body,
+                _headers={"X-Per-Request-Header": "per-request-value"},
+            )
+            self.assertIsInstance(api_response, CheckResponse)
+            self.assertTrue(api_response.allowed)
+            # Make sure both headers are present in the request
+            expected_headers = urllib3.response.HTTPHeaderDict(
+                {
+                    "Accept": "application/json",
+                    "Content-Type": "application/json",
+                    "User-Agent": "openfga-sdk python/0.9.6",
+                    "X-Default-Header": "default-value",  # Default header preserved
+                    "X-Per-Request-Header": "per-request-value",  # Per-request header added
+                }
+            )
+            mock_request.assert_called_once_with(
+                "POST",
+                "http://api.fga.example/stores/01H0H015178Y2V4CX10C2KGHF4/check",
+                headers=expected_headers,
+                query_params=[],
+                post_params=[],
+                body={
+                    "tuple_key": {
+                        "object": "document:2021-budget",
+                        "relation": "reader",
+                        "user": "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
+                    }
+                },
+                _preload_content=ANY,
+                _request_timeout=None,
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

test/client/client_test.py

@@ -21,7 +21,7 @@
 
 from openfga_sdk import rest
 from openfga_sdk.client import ClientConfiguration
-from openfga_sdk.client.client import OpenFgaClient
+from openfga_sdk.client.client import OpenFgaClient, set_heading_if_not_set
 from openfga_sdk.client.models.assertion import ClientAssertion
 from openfga_sdk.client.models.batch_check_item import ClientBatchCheckItem
 from openfga_sdk.client.models.batch_check_request import ClientBatchCheckRequest
@@ -3273,3 +3273,100 @@ def test_configuration_authorization_model_id_invalid(self):
             authorization_model_id="abcd",
         )
         self.assertRaises(FgaValidationException, configuration.is_valid)
+
+    def test_set_heading_if_not_set_when_none_provided(self):
+        """Should set header when no options provided"""
+        result = set_heading_if_not_set(None, "X-Test-Header", "default-value")
+
+        self.assertIsNotNone(result)
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_when_empty_options_provided(self):
+        """Should set header when empty options dict provided"""
+        result = set_heading_if_not_set({}, "X-Test-Header", "default-value")
+
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_when_no_headers_in_options(self):
+        """Should set header when options dict has no headers key"""
+        options = {"page_size": 10}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+        self.assertEqual(result["page_size"], 10)
+
+    def test_set_heading_if_not_set_when_headers_empty(self):
+        """Should set header when headers dict is empty"""
+        options = {"headers": {}}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_does_not_override_existing_custom_header(self):
+        """Should NOT override when custom header already exists - this is the critical test for the bug fix"""
+        options = {"headers": {"X-Test-Header": "custom-value"}}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        # Custom header should be preserved, NOT overridden by default
+        self.assertEqual(result["headers"]["X-Test-Header"], "custom-value")
+
+    def test_set_heading_if_not_set_preserves_other_headers_when_setting_new_header(
+        self,
+    ):
+        """Should preserve existing headers when setting a new one"""
+        options = {"headers": {"X-Existing-Header": "existing-value"}}
+        result = set_heading_if_not_set(options, "X-New-Header", "new-value")
+
+        self.assertEqual(result["headers"]["X-Existing-Header"], "existing-value")
+        self.assertEqual(result["headers"]["X-New-Header"], "new-value")
+
+    def test_set_heading_if_not_set_handles_integer_header_values(self):
+        """Should not override existing integer header values"""
+        options = {"headers": {"X-Retry-Count": 5}}
+        result = set_heading_if_not_set(options, "X-Retry-Count", 1)
+
+        # Existing integer value should be preserved
+        self.assertEqual(result["headers"]["X-Retry-Count"], 5)
+
+    def test_set_heading_if_not_set_handles_non_dict_headers_value(self):
+        """Should convert non-dict headers value to dict"""
+        options = {"headers": "invalid"}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertIsInstance(result["headers"], dict)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_does_not_mutate_when_header_exists(self):
+        """Should return same dict when header already exists"""
+        options = {"headers": {"X-Test-Header": "custom-value"}}
+        original_value = options["headers"]["X-Test-Header"]
+
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        # Should return the same modified dict
+        self.assertIs(result, options)
+        # Value should not have changed
+        self.assertEqual(result["headers"]["X-Test-Header"], original_value)
+
+    def test_set_heading_if_not_set_multiple_headers_with_mixed_states(self):
+        """Should handle multiple headers, some existing and some new"""
+        options = {
+            "headers": {
+                "X-Custom-Header": "custom-value",
+                "X-Another-Header": "another-value",
+            }
+        }
+
+        # Try to set a custom header (should not override)
+        result = set_heading_if_not_set(options, "X-Custom-Header", "default-value")
+        self.assertEqual(result["headers"]["X-Custom-Header"], "custom-value")
+
+        # Try to set a new header (should be added)
+        result = set_heading_if_not_set(result, "X-New-Header", "new-value")
+        self.assertEqual(result["headers"]["X-New-Header"], "new-value")
+
+        # Original headers should still exist
+        self.assertEqual(result["headers"]["X-Another-Header"], "another-value")

test/sync/client/client_test.py

@@ -85,7 +85,7 @@
     WriteAuthorizationModelResponse,
 )
 from openfga_sdk.sync import rest
-from openfga_sdk.sync.client.client import OpenFgaClient
+from openfga_sdk.sync.client.client import OpenFgaClient, set_heading_if_not_set
 
 
 store_id = "01YCP46JKYM8FJCQ37NMBYHE5X"
@@ -3275,3 +3275,100 @@ def test_configuration_authorization_model_id_invalid(self):
             authorization_model_id="abcd",
         )
         self.assertRaises(FgaValidationException, configuration.is_valid)
+
+    def test_set_heading_if_not_set_when_none_provided(self):
+        """Should set header when no options provided"""
+        result = set_heading_if_not_set(None, "X-Test-Header", "default-value")
+
+        self.assertIsNotNone(result)
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_when_empty_options_provided(self):
+        """Should set header when empty options dict provided"""
+        result = set_heading_if_not_set({}, "X-Test-Header", "default-value")
+
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_when_no_headers_in_options(self):
+        """Should set header when options dict has no headers key"""
+        options = {"page_size": 10}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertIn("headers", result)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+        self.assertEqual(result["page_size"], 10)
+
+    def test_set_heading_if_not_set_when_headers_empty(self):
+        """Should set header when headers dict is empty"""
+        options = {"headers": {}}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_does_not_override_existing_custom_header(self):
+        """Should NOT override when custom header already exists - this is the critical test for the bug fix"""
+        options = {"headers": {"X-Test-Header": "custom-value"}}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        # Custom header should be preserved, NOT overridden by default
+        self.assertEqual(result["headers"]["X-Test-Header"], "custom-value")
+
+    def test_set_heading_if_not_set_preserves_other_headers_when_setting_new_header(
+        self,
+    ):
+        """Should preserve existing headers when setting a new one"""
+        options = {"headers": {"X-Existing-Header": "existing-value"}}
+        result = set_heading_if_not_set(options, "X-New-Header", "new-value")
+
+        self.assertEqual(result["headers"]["X-Existing-Header"], "existing-value")
+        self.assertEqual(result["headers"]["X-New-Header"], "new-value")
+
+    def test_set_heading_if_not_set_handles_integer_header_values(self):
+        """Should not override existing integer header values"""
+        options = {"headers": {"X-Retry-Count": 5}}
+        result = set_heading_if_not_set(options, "X-Retry-Count", 1)
+
+        # Existing integer value should be preserved
+        self.assertEqual(result["headers"]["X-Retry-Count"], 5)
+
+    def test_set_heading_if_not_set_handles_non_dict_headers_value(self):
+        """Should convert non-dict headers value to dict"""
+        options = {"headers": "invalid"}
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        self.assertIsInstance(result["headers"], dict)
+        self.assertEqual(result["headers"]["X-Test-Header"], "default-value")
+
+    def test_set_heading_if_not_set_does_not_mutate_when_header_exists(self):
+        """Should return same dict when header already exists"""
+        options = {"headers": {"X-Test-Header": "custom-value"}}
+        original_value = options["headers"]["X-Test-Header"]
+
+        result = set_heading_if_not_set(options, "X-Test-Header", "default-value")
+
+        # Should return the same modified dict
+        self.assertIs(result, options)
+        # Value should not have changed
+        self.assertEqual(result["headers"]["X-Test-Header"], original_value)
+
+    def test_set_heading_if_not_set_multiple_headers_with_mixed_states(self):
+        """Should handle multiple headers, some existing and some new"""
+        options = {
+            "headers": {
+                "X-Custom-Header": "custom-value",
+                "X-Another-Header": "another-value",
+            }
+        }
+
+        # Try to set a custom header (should not override)
+        result = set_heading_if_not_set(options, "X-Custom-Header", "default-value")
+        self.assertEqual(result["headers"]["X-Custom-Header"], "custom-value")
+
+        # Try to set a new header (should be added)
+        result = set_heading_if_not_set(result, "X-New-Header", "new-value")
+        self.assertEqual(result["headers"]["X-New-Header"], "new-value")
+
+        # Original headers should still exist
+        self.assertEqual(result["headers"]["X-Another-Header"], "another-value")