fix: streaming stuff

Author: SoulPancake

CHANGELOG.md

@@ -6,7 +6,7 @@
 
 ### Breaking Changes
 
-- The `_return_http_data_only`, `_request_auth`, and `_preload_content` kwargs have been removed from all `OpenFgaApi` and `SyncOpenFgaApi` endpoint methods. These were internal implementation details not intended for external use. `_return_http_data_only` is now hardcoded to `True` internally, meaning all endpoint methods always return the deserialized response object directly. Users relying on `_with_http_info` methods returning a `(data, status, headers)` tuple should use `execute_api_request` instead.
+- The `_return_http_data_only`, `_preload_content`, `_request_auth`, `async_req`, and `_request_timeout` kwargs have been removed from all `OpenFgaApi` and `SyncOpenFgaApi` endpoint methods. These were internal implementation details not intended for external use. `_return_http_data_only` is now hardcoded to `True`; all endpoint methods return the deserialized response object directly. Users relying on `_with_http_info` methods returning a `(data, status, headers)` tuple should use `execute_api_request` instead.
 
 ### [0.9.9](https://github.com/openfga/python-sdk/compare/v0.9.8...v0.9.9) (2025-12-09)
 - feat: improve error messaging (#245)

README.md

@@ -1265,7 +1265,7 @@ response = await fga_client.write_assertions(body, options)
 
 In certain cases you may want to call other APIs not yet wrapped by the SDK. You can do so by using the `execute_api_request` method available on the `OpenFgaClient`. It allows you to make raw HTTP calls to any OpenFGA endpoint by specifying the HTTP method, path, body, query parameters, and path parameters, while still honoring the client configuration (authentication, telemetry, retries, and error handling).
 
-For streaming endpoints, use `execute_streamed_api_request` instead.
+For streaming endpoints (e.g. `streamed-list-objects`), use `execute_streamed_api_request` instead. It returns an `AsyncIterator` (or `Iterator` in the sync client) that yields one parsed JSON object per chunk.
 
 This is useful when:
 - You want to call a new endpoint that is not yet supported by the SDK
@@ -1315,6 +1315,27 @@ stores = stores_response.json()
 print("Stores:", stores)
 ```
 
+#### Example: Calling a Streaming Endpoint
+
+```python
+# Stream objects visible to a user
+async for chunk in fga_client.execute_streamed_api_request(
+    operation_name="StreamedListObjects",
+    method="POST",
+    path="/stores/{store_id}/streamed-list-objects",
+    path_params={"store_id": FGA_STORE_ID},
+    body={
+        "type": "document",
+        "relation": "viewer",
+        "user": "user:anne",
+        "authorization_model_id": FGA_MODEL_ID,
+    },
+):
+    # Each chunk has the shape {"result": {"object": "..."}} or {"error": {...}}
+    if "result" in chunk:
+        print(chunk["result"]["object"])  # e.g. "document:roadmap"
+```
+
 #### Example: Using Path Parameters
 
 Path parameters are specified in the path using `{param_name}` syntax and must all be provided explicitly via `path_params` (URL-encoded automatically):

example/execute-api-request/execute_api_request_example.py

@@ -242,6 +242,27 @@ async def main():
         assert raw.status == 200
         print(f"   ✅ custom headers accepted (status {raw.status})")
 
+        print("9. StreamedListObjects (POST /stores/{store_id}/streamed-list-objects)")
+        chunks = []
+        async for chunk in fga_client.execute_streamed_api_request(
+            operation_name="StreamedListObjects",
+            method="POST",
+            path="/stores/{store_id}/streamed-list-objects",
+            path_params={"store_id": store.id},
+            body={
+                "type": "document",
+                "relation": "viewer",
+                "user": "user:anne",
+                "authorization_model_id": auth_model_id,
+            },
+        ):
+            chunks.append(chunk)
+        assert len(chunks) >= 1, f"Expected at least 1 chunk, got {len(chunks)}"
+        # Each chunk has the shape {"result": {"object": "..."}} or {"error": {...}}
+        objects = [c["result"]["object"] for c in chunks if "result" in c]
+        assert "document:roadmap" in objects, f"Expected document:roadmap in {objects}"
+        print(f"   ✅ {len(chunks)} chunks, objects={objects}")
+
         print("\n=== Cleanup ===")
         await fga_client.delete_store()
         print(f"Deleted test store: {store.id}")

openfga_sdk/__init__.py

@@ -2,6 +2,7 @@
 from openfga_sdk.api_client import ApiClient
 from openfga_sdk.client.client import OpenFgaClient
 from openfga_sdk.client.configuration import ClientConfiguration
+from openfga_sdk.client.models.raw_response import RawResponse
 from openfga_sdk.configuration import Configuration
 from openfga_sdk.constants import SDK_VERSION
 from openfga_sdk.exceptions import (
@@ -136,6 +137,7 @@
 __all__ = [
     "OpenFgaClient",
     "ClientConfiguration",
+    "RawResponse",
     "OpenFgaApi",
     "ApiClient",
     "Configuration",

openfga_sdk/api/open_fga_api.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any
 
 from openfga_sdk.api_client import ApiClient
@@ -148,7 +149,7 @@ async def execute_api_request(
         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, int | str | dict[str, int | str]] | None = None,
+        options: dict[str, Any] | None = None,
     ) -> RawResponse:
         """
         Execute an arbitrary HTTP request to any OpenFGA API endpoint.
@@ -164,7 +165,7 @@ async def execute_api_request(
         :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 (headers, retry_params)
+        :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
         :return: RawResponse with status, headers, and body
         """
         return await self._execute_api_request_internal(
@@ -176,7 +177,6 @@ async def execute_api_request(
             query_params=query_params,
             headers=headers,
             options=options,
-            streaming=False,
         )
 
     async def execute_streamed_api_request(
@@ -189,25 +189,73 @@ async def execute_streamed_api_request(
         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, int | str | dict[str, int | str]] | None = None,
-    ) -> RawResponse:
+        options: dict[str, Any] | None = None,
+    ) -> AsyncIterator[dict[str, Any]]:
         """
         Execute an arbitrary HTTP request to a streaming OpenFGA API endpoint.
 
-        Same interface as execute_api_request but for streaming endpoints.
-        See execute_api_request for full parameter documentation.
+        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)}
         """
-        return await self._execute_api_request_internal(
+        from openfga_sdk.client.execute_api_request_builder import (
+            ExecuteApiRequestBuilder,
+        )
+
+        builder = ExecuteApiRequestBuilder(
             operation_name=operation_name,
             method=method,
             path=path,
             path_params=path_params,
             body=body,
             query_params=query_params,
             headers=headers,
-            options=options,
-            streaming=True,
         )
+        builder.validate()
+
+        resource_path = builder.build_path()
+        query_params_list = builder.build_query_params_list()
+        final_headers = builder.build_headers()
+
+        retry_params = options.get("retry_params") if options else None
+
+        telemetry_attributes: dict[TelemetryAttribute, str | bool | int | float] = {
+            TelemetryAttributes.fga_client_request_method: operation_name.lower(),
+        }
+        if self.api_client.get_store_id():
+            telemetry_attributes[TelemetryAttributes.fga_client_request_store_id] = (
+                self.api_client.get_store_id()
+            )
+
+        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={},
+            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,
+        )
+
+        async for chunk in stream:
+            yield chunk
 
     async def _execute_api_request_internal(
         self,
@@ -219,10 +267,9 @@ async def _execute_api_request_internal(
         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, int | str | dict[str, int | str]] | None = None,
-        streaming: bool = False,
+        options: dict[str, Any] | None = None,
     ) -> RawResponse:
-        """Shared implementation for execute_api_request and execute_streamed_api_request."""
+        """Implementation for execute_api_request."""
         from openfga_sdk.client.execute_api_request_builder import (
             ExecuteApiRequestBuilder,
             ResponseParser,
@@ -242,15 +289,9 @@ async def _execute_api_request_internal(
 
         resource_path = builder.build_path()
         query_params_list = builder.build_query_params_list()
+        final_headers = builder.build_headers()
 
-        options_headers = None
-        if options and isinstance(options.get("headers"), dict):
-            options_headers = options["headers"]
-        final_headers = builder.build_headers(options_headers)
-
-        retry_params = None
-        if options and options.get("retry_params"):
-            retry_params = options["retry_params"]
+        retry_params = options.get("retry_params") if options else None
 
         telemetry_attributes: dict[TelemetryAttribute, str | bool | int | float] = {
             TelemetryAttributes.fga_client_request_method: operation_name.lower(),
@@ -273,7 +314,7 @@ async def _execute_api_request_internal(
             _retry_params=retry_params,
             _oauth2_client=self._oauth2_client,
             _telemetry_attributes=telemetry_attributes,
-            _streaming=streaming,
+            _streaming=False,
         )
 
         rest_response = getattr(self.api_client, "last_response", None)

openfga_sdk/client/client.py

@@ -1,6 +1,7 @@
 import asyncio
 import uuid
 
+from collections.abc import AsyncIterator
 from typing import Any
 
 from openfga_sdk.api.open_fga_api import OpenFgaApi
@@ -1113,7 +1114,7 @@ async def execute_api_request(
         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, int | str | dict[str, int | str]] | None = None,
+        options: dict[str, Any] | None = None,
     ) -> RawResponse:
         """
         Execute an arbitrary HTTP request to any OpenFGA API endpoint.
@@ -1129,7 +1130,7 @@ async def execute_api_request(
         :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 (headers, retry_params)
+        :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
         :return: RawResponse with status, headers, and body
         """
         return await self._api.execute_api_request(
@@ -1153,15 +1154,17 @@ async def execute_streamed_api_request(
         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, int | str | dict[str, int | str]] | None = None,
-    ) -> RawResponse:
+        options: dict[str, Any] | None = None,
+    ) -> AsyncIterator[dict[str, Any]]:
         """
         Execute an arbitrary HTTP request to a streaming OpenFGA API endpoint.
 
         Same interface as execute_api_request but for streaming endpoints.
         See execute_api_request for full parameter documentation.
+
+        :return: AsyncIterator yielding parsed JSON chunks from the streaming response.
         """
-        return await self._api.execute_streamed_api_request(
+        async for chunk in self._api.execute_streamed_api_request(
             operation_name=operation_name,
             method=method,
             path=path,
@@ -1170,4 +1173,5 @@ async def execute_streamed_api_request(
             query_params=query_params,
             headers=headers,
             options=options,
-        )
+        ):
+            yield chunk

openfga_sdk/sync/client/client.py

@@ -1,5 +1,6 @@
 import uuid
 
+from collections.abc import Iterator
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
@@ -1111,7 +1112,7 @@ def execute_api_request(
         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, int | str | dict[str, int | str]] | None = None,
+        options: dict[str, Any] | None = None,
     ) -> RawResponse:
         """
         Execute an arbitrary HTTP request to any OpenFGA API endpoint.
@@ -1127,7 +1128,7 @@ def execute_api_request(
         :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 (headers, retry_params)
+        :param options: Extra options e.g. {"retry_params": RetryParams(max_retry=3)}
         :return: RawResponse with status, headers, and body
         """
         return self._api.execute_api_request(
@@ -1151,15 +1152,17 @@ def execute_streamed_api_request(
         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, int | str | dict[str, int | str]] | None = None,
-    ) -> RawResponse:
+        options: dict[str, Any] | None = None,
+    ) -> Iterator[dict[str, Any]]:
         """
         Execute an arbitrary HTTP request to a streaming OpenFGA API endpoint.
 
         Same interface as execute_api_request but for streaming endpoints.
         See execute_api_request for full parameter documentation.
+
+        :return: Iterator yielding parsed JSON chunks from the streaming response.
         """
-        return self._api.execute_streamed_api_request(
+        yield from self._api.execute_streamed_api_request(
             operation_name=operation_name,
             method=method,
             path=path,