fix: refactor

Author: SoulPancake

openfga_sdk/api/open_fga_api.py

@@ -88,9 +88,8 @@ async def _execute(
     ) -> Any:
         """Shared executor for all API endpoint methods.
 
-        Delegates to _execute_api_request_internal (or
-        _execute_streamed_api_request_internal when _streaming=True) so all
-        API calls share one unified code path.
+        Delegates to the public execute_api_request / execute_streamed_api_request
+        so all API calls share one unified code path.
         """
         if options is None:
             options = {}
@@ -118,7 +117,7 @@ async def _execute(
         )
 
         if options.get("_streaming", False):
-            return self._execute_streamed_api_request_internal(
+            return self.execute_streamed_api_request(
                 operation_name=operation_name,
                 method=method,
                 path=path,
@@ -128,9 +127,10 @@ async def _execute(
                 options=options,
                 response_types_map=response_types_map,
                 telemetry_attributes=telemetry_attributes,
+                _skip_builder=True,
             )
 
-        return await self._execute_api_request_internal(
+        return await self.execute_api_request(
             operation_name=operation_name,
             method=method,
             path=path,
@@ -151,16 +151,22 @@ async def execute_api_request(
         path: str,
         path_params: dict[str, str] | None = None,
         body: dict[str, Any] | list[Any] | str | bytes | None = None,
-        query_params: dict[str, str | int | list[str | int]] | None = None,
+        query_params: dict[str, str | int | list[str | int]] | list | None = None,
         headers: dict[str, str] | None = None,
         options: dict[str, Any] | None = None,
-    ) -> RawResponse:
+        response_types_map: dict | None = None,
+        telemetry_attributes: dict | None = None,
+        _skip_builder: bool = False,
+    ) -> Any:
         """
         Execute an arbitrary HTTP request to any OpenFGA API endpoint.
 
         Useful for calling endpoints not yet wrapped by the SDK while
         still getting authentication, retries, and error handling.
 
+        When called directly by users, returns a RawResponse.
+        When called internally with response_types_map, returns a deserialized typed model.
+
         :param operation_name: Operation name for telemetry (e.g., "CustomCheck")
         :param method: HTTP method (GET, POST, PUT, DELETE, PATCH)
         :param path: API path, e.g. "/stores/{store_id}/my-endpoint".
@@ -170,92 +176,26 @@ async def execute_api_request(
         :param query_params: Query string parameters
         :param headers: Custom headers (SDK enforces Content-Type and Accept)
         :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
-        :return: RawResponse with status, headers, and body
+        :param response_types_map: When provided, merge with common error types and
+            return a deserialized typed model. When None, return RawResponse.
+        :param telemetry_attributes: When provided, use as-is for telemetry.
+            When None, build minimal defaults.
+        :param _skip_builder: When True, path/headers/query_params are used as-is
+            (already resolved by _execute). When False (default), run through
+            ExecuteApiRequestBuilder.
+        :return: RawResponse (public callers) or deserialized model (internal callers)
         """
-        return await self._execute_api_request_internal(
-            operation_name=operation_name,
-            method=method,
-            path=path,
-            path_params=path_params,
-            body=body,
-            query_params=query_params,
-            headers=headers,
-            options=options,
+        from openfga_sdk.client.execute_api_request_builder import (
+            ExecuteApiRequestBuilder,
+            ResponseParser,
         )
+        from openfga_sdk.client.models.raw_response import RawResponse
 
-    async def execute_streamed_api_request(
-        self,
-        *,
-        operation_name: str,
-        method: str,
-        path: str,
-        path_params: dict[str, str] | None = None,
-        body: dict[str, Any] | list[Any] | str | bytes | None = None,
-        query_params: dict[str, str | int | list[str | int]] | None = None,
-        headers: dict[str, str] | None = None,
-        options: dict[str, Any] | None = None,
-    ) -> AsyncIterator[dict[str, Any]]:
-        """
-        Execute an arbitrary HTTP request to a streaming OpenFGA API endpoint.
-
-        Yields parsed JSON objects as they arrive. Use with async for:
-
-            async for chunk in api.execute_streamed_api_request(...):
-                process(chunk)
-
-        :param operation_name: Operation name for telemetry (e.g., "StreamedListObjects")
-        :param method: HTTP method (GET, POST, PUT, DELETE, PATCH)
-        :param path: API path, e.g. "/stores/{store_id}/streamed-list-objects".
-        :param path_params: Path parameter substitutions (URL-encoded automatically).
-            All path parameters, including store_id, must be provided explicitly.
-        :param body: Request body for POST/PUT/PATCH
-        :param query_params: Query string parameters
-        :param headers: Custom headers (SDK enforces Content-Type and Accept)
-        :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
-        """
-        async for chunk in self._execute_streamed_api_request_internal(
-            operation_name=operation_name,
-            method=method,
-            path=path,
-            path_params=path_params,
-            body=body,
-            query_params=query_params,
-            headers=headers,
-            options=options,
-        ):
-            yield chunk
-
-    async def _execute_streamed_api_request_internal(
-        self,
-        *,
-        operation_name: str,
-        method: str,
-        path: str,
-        path_params: dict[str, str] | None = None,
-        body: dict[str, Any] | list[Any] | str | bytes | None = None,
-        query_params=None,
-        headers=None,
-        options: dict[str, Any] | None = None,
-        response_types_map: dict | None = None,
-        telemetry_attributes: dict | None = None,
-        _skip_builder: bool = False,
-    ) -> AsyncIterator[dict[str, Any]]:
-        """Implementation for streaming API requests.
-
-        When _skip_builder=True, query_params must already be a list of tuples
-        and headers must already be a fully-built dict (used when called from _execute).
-        When response_types_map is provided, it is merged with _COMMON_ERROR_RESPONSE_TYPES.
-        When telemetry_attributes is provided, it is used as-is instead of building minimal defaults.
-        """
         if _skip_builder:
             resource_path = path
             query_params_list = query_params if query_params is not None else []
             final_headers = headers or {}
         else:
-            from openfga_sdk.client.execute_api_request_builder import (
-                ExecuteApiRequestBuilder,
-            )
-
             builder = ExecuteApiRequestBuilder(
                 operation_name=operation_name,
                 method=method,
@@ -272,6 +212,8 @@ async def _execute_streamed_api_request_internal(
             final_headers = builder.build_headers()
 
         retry_params = options.get("retry_params") if options else None
+        request_timeout = options.get("_request_timeout") if options else None
+        async_req = options.get("async_req") if options else None
 
         if telemetry_attributes is None:
             telemetry_attributes = {
@@ -287,62 +229,103 @@ async def _execute_streamed_api_request_internal(
                 **self._COMMON_ERROR_RESPONSE_TYPES,
                 **response_types_map,
             }
-        else:
-            merged_response_types_map = {}
+            return await self.api_client.call_api(
+                resource_path=resource_path,
+                method=method.upper(),
+                path_params={},
+                query_params=query_params_list if query_params_list else [],
+                header_params=final_headers,
+                body=body,
+                post_params=[],
+                files={},
+                response_types_map=merged_response_types_map,
+                auth_settings=[],
+                async_req=async_req,
+                _return_http_data_only=True,
+                _preload_content=True,
+                _request_timeout=request_timeout,
+                _retry_params=retry_params,
+                collection_formats={},
+                _oauth2_client=self._oauth2_client,
+                _telemetry_attributes=telemetry_attributes,
+                _streaming=False,
+            )
 
-        stream = await self.api_client.call_api(
+        # No response_types_map: public caller path → return RawResponse
+        await self.api_client.call_api(
             resource_path=resource_path,
             method=method.upper(),
             query_params=query_params_list if query_params_list else None,
             header_params=final_headers,
             body=body,
-            response_types_map=merged_response_types_map,
+            response_types_map={},
             auth_settings=[],
             _return_http_data_only=True,
             _preload_content=True,
             _retry_params=retry_params,
             _oauth2_client=self._oauth2_client,
             _telemetry_attributes=telemetry_attributes,
-            _streaming=True,
+            _streaming=False,
         )
 
-        async for chunk in stream:
-            yield chunk
+        rest_response = getattr(self.api_client, "last_response", None)
+        if rest_response is None:
+            raise RuntimeError(
+                f"No response for {method.upper()} {resource_path} "
+                f"(operation: {operation_name})"
+            )
+
+        return RawResponse(
+            status=rest_response.status,
+            headers=dict(rest_response.getheaders()),
+            body=ResponseParser.parse_body(rest_response.data),
+        )
 
-    async def _execute_api_request_internal(
+    async def execute_streamed_api_request(
         self,
         *,
         operation_name: str,
         method: str,
         path: str,
         path_params: dict[str, str] | None = None,
         body: dict[str, Any] | list[Any] | str | bytes | None = None,
-        query_params=None,
-        headers=None,
+        query_params: dict[str, str | int | list[str | int]] | list | None = None,
+        headers: dict[str, str] | None = None,
         options: dict[str, Any] | None = None,
         response_types_map: dict | None = None,
         telemetry_attributes: dict | None = None,
         _skip_builder: bool = False,
-    ) -> Any:
-        """Implementation for execute_api_request (and _execute delegation).
-
-        When _skip_builder=True, query_params must already be a list of tuples
-        and headers must already be a fully-built dict (used when called from _execute).
-        When response_types_map is provided, it is merged with _COMMON_ERROR_RESPONSE_TYPES
-        and the deserialized typed result from call_api is returned directly.
-        When telemetry_attributes is provided, it is used as-is instead of building minimal defaults.
+    ) -> AsyncIterator[dict[str, Any]]:
         """
-        from openfga_sdk.client.execute_api_request_builder import (
-            ExecuteApiRequestBuilder,
-            ResponseParser,
-        )
-        from openfga_sdk.client.models.raw_response import RawResponse
+        Execute an arbitrary HTTP request to a streaming OpenFGA API endpoint.
+
+        Yields parsed JSON objects as they arrive. Use with async for:
+
+            async for chunk in api.execute_streamed_api_request(...):
+                process(chunk)
 
+        :param operation_name: Operation name for telemetry (e.g., "StreamedListObjects")
+        :param method: HTTP method (GET, POST, PUT, DELETE, PATCH)
+        :param path: API path, e.g. "/stores/{store_id}/streamed-list-objects".
+        :param path_params: Path parameter substitutions (URL-encoded automatically).
+            All path parameters, including store_id, must be provided explicitly.
+        :param body: Request body for POST/PUT/PATCH
+        :param query_params: Query string parameters
+        :param headers: Custom headers (SDK enforces Content-Type and Accept)
+        :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
+        :param response_types_map: When provided, merge with common error types.
+        :param telemetry_attributes: When provided, use as-is for telemetry.
+        :param _skip_builder: When True, path/headers/query_params are used as-is.
+        """
         if _skip_builder:
             resource_path = path
             query_params_list = query_params if query_params is not None else []
             final_headers = headers or {}
         else:
+            from openfga_sdk.client.execute_api_request_builder import (
+                ExecuteApiRequestBuilder,
+            )
+
             builder = ExecuteApiRequestBuilder(
                 operation_name=operation_name,
                 method=method,
@@ -359,8 +342,6 @@ async def _execute_api_request_internal(
             final_headers = builder.build_headers()
 
         retry_params = options.get("retry_params") if options else None
-        request_timeout = options.get("_request_timeout") if options else None
-        async_req = options.get("async_req") if options else None
 
         if telemetry_attributes is None:
             telemetry_attributes = {
@@ -376,58 +357,27 @@ async def _execute_api_request_internal(
                 **self._COMMON_ERROR_RESPONSE_TYPES,
                 **response_types_map,
             }
-            # call_api returns a deserialized typed model when response_types_map is non-empty
-            return await self.api_client.call_api(
-                resource_path=resource_path,
-                method=method.upper(),
-                path_params={},
-                query_params=query_params_list if query_params_list else [],
-                header_params=final_headers,
-                body=body,
-                post_params=[],
-                files={},
-                response_types_map=merged_response_types_map,
-                auth_settings=[],
-                async_req=async_req,
-                _return_http_data_only=True,
-                _preload_content=True,
-                _request_timeout=request_timeout,
-                _retry_params=retry_params,
-                collection_formats={},
-                _oauth2_client=self._oauth2_client,
-                _telemetry_attributes=telemetry_attributes,
-                _streaming=False,
-            )
+        else:
+            merged_response_types_map = {}
 
-        # No response_types_map: public execute_api_request path → return RawResponse
-        await self.api_client.call_api(
+        stream = await self.api_client.call_api(
             resource_path=resource_path,
             method=method.upper(),
             query_params=query_params_list if query_params_list else None,
             header_params=final_headers,
             body=body,
-            response_types_map={},
+            response_types_map=merged_response_types_map,
             auth_settings=[],
             _return_http_data_only=True,
             _preload_content=True,
             _retry_params=retry_params,
             _oauth2_client=self._oauth2_client,
             _telemetry_attributes=telemetry_attributes,
-            _streaming=False,
+            _streaming=True,
         )
 
-        rest_response = getattr(self.api_client, "last_response", None)
-        if rest_response is None:
-            raise RuntimeError(
-                f"No response for {method.upper()} {resource_path} "
-                f"(operation: {operation_name})"
-            )
-
-        return RawResponse(
-            status=rest_response.status,
-            headers=dict(rest_response.getheaders()),
-            body=ResponseParser.parse_body(rest_response.data),
-        )
+        async for chunk in stream:
+            yield chunk
 
     async def batch_check(self, body, **kwargs):
         """Send a list of `check` operations in a single request