feat: add execute_api_request method for calling arbitrary endpoints by kcbiradar · Pull Request #252 · openfga/python-sdk

Documentation Updates

2 document(s) were updated by changes in this PR:

Custom HTTP Headers Support
View Changes
@@ -191,6 +191,87 @@
 #### Interface Changes
 - `headers` dictionary in `ClientConfiguration` for global headers.
 - `headers` dictionary in the `options` parameter for per-request headers.
+
+#### Calling Other Endpoints with execute_api_request
+The Python SDK provides `execute_api_request` and `execute_streamed_api_request` methods for making raw HTTP calls to any OpenFGA endpoint, including endpoints not yet wrapped by the SDK. These methods allow you to specify the HTTP method, path, body, query parameters, and path parameters, while still honoring the client configuration (authentication, telemetry, retries, and error handling).
+
+This is useful when:
+- You want to call a new endpoint that is not yet supported by the SDK
+- You are using an earlier version of the SDK that doesn't yet support a particular endpoint
+- You have a custom endpoint deployed that extends the OpenFGA API
+
+**Basic POST Request:**
+```python
+# Call a custom endpoint using path parameters
+response = await fga_client.execute_api_request(
+    operation_name="CustomEndpoint",        # For telemetry/logging
+    method="POST",
+    path="/stores/{store_id}/custom-endpoint",
+    path_params={"store_id": FGA_STORE_ID},
+    body={
+        "user": "user:bob",
+        "action": "custom_action",
+        "resource": "resource:123",
+    },
+    query_params={
+        "page_size": 20,
+    },
+)
+
+# Access the response data
+if response.status == 200:
+    result = response.json()
+    print(f"Response: {result}")
+```
+
+**GET Request with Query Parameters:**
+```python
+# Get a list of stores with query parameters
+stores_response = await fga_client.execute_api_request(
+    operation_name="ListStores",
+    method="GET",
+    path="/stores",
+    query_params={
+        "page_size": 10,
+        "continuation_token": "eyJwayI6...",
+    },
+)
+
+stores = stores_response.json()
+print("Stores:", stores)
+```
+
+**Streaming Endpoint Example:**
+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.
+
+```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"
+```
+
+**Response Object (RawResponse):**
+The `RawResponse` object provides:
+- `status` - HTTP status code
+- `headers` - Response headers dictionary
+- `body` - Response body (parsed as dict if JSON, otherwise str or bytes)
+- `json()` - Returns the parsed JSON body if available
+- `text()` - Returns the body as a string
+
+For more examples, see the [Python SDK README](https://github.com/openfga/python-sdk#calling-other-endpoints).
 
 ---
 
StreamedListObjects Feature Overview
View Changes
@@ -52,6 +52,26 @@
     print(response.object)
 ```
 You can pass options for authorization model ID, consistency, headers, and retry parameters. The method yields `StreamedListObjectsResponse` objects as soon as they are available ([API details](https://github.com/openfga/python-sdk/blob/fd04bc4b8525e536208e2091dce16edf2f220250/openfga_sdk/client/client.py#L929-L967)).
+
+Additionally, the Python SDK provides `execute_api_request` and `execute_streamed_api_request` methods on `OpenFgaClient` that can be used to call arbitrary API endpoints that don't yet have dedicated SDK methods. Use `execute_api_request` for regular endpoints and `execute_streamed_api_request` for streaming endpoints like `/streamed-list-objects`. These methods act as an escape hatch while still providing SDK benefits like authentication, configuration, retries, and error handling.
+
+**Example using execute_streamed_api_request:**
+```python
+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,
+    },
+):
+    if "result" in chunk:
+        print(chunk["result"]["object"])
+```
 
 ### Go SDK
 The Go SDK provides a public `StreamedListObjects` method on `OpenFgaClient`, which streams results using channels. This method allows you to retrieve all objects of a given type that a user has access to, without the 1000-object pagination limit.

How did I do? Any feedback?  Join Discord