feat: add raw requestor for calling arbitrary endpoints

README.md

@@ -48,6 +48,7 @@ This is an autogenerated python SDK for OpenFGA. It provides a wrapper around th
       - [Read Assertions](#read-assertions)
       - [Write Assertions](#write-assertions)
   - [Retries](#retries)
+  - [Calling Other Endpoints](#calling-other-endpoints)
   - [API Endpoints](#api-endpoints)
   - [Models](#models)
   - [OpenTelemetry](#opentelemetry)
@@ -1260,6 +1261,137 @@ body = [ClientAssertion(
 response = await fga_client.write_assertions(body, options)
 ```
 
+### Calling Other Endpoints
+
+In certain cases you may want to call other APIs not yet wrapped by the SDK. You can do so by using the `raw_request` method available on the `OpenFgaClient`. The `raw_request` method allows you to make raw HTTP calls to any OpenFGA endpoint by specifying the operation name, HTTP method, path, parameters, body, and headers, 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
+
+In all cases, you initialize the SDK the same way as usual, and then call `raw_request` on the `fga_client` instance.
+
+```python
+from openfga_sdk import ClientConfiguration, OpenFgaClient
+
+configuration = ClientConfiguration(
+    api_url=FGA_API_URL,
+    store_id=FGA_STORE_ID,
+)
+
+async with OpenFgaClient(configuration) as fga_client:
+    request_body = {
+        "user": "user:bob",
+        "action": "custom_action",
+        "resource": "resource:123",
+    }
+
+    response = await fga_client.raw_request(
+        operation_name="CustomEndpoint",
+        method="POST",
+        path="/stores/{store_id}/custom-endpoint",
+        path_params={"store_id": FGA_STORE_ID},
+        query_params={"page_size": "20"},
+        body=request_body,
+        headers={"X-Experimental-Feature": "enabled"},
+    )
+```
+
+#### Example: Calling a new "Custom Endpoint" endpoint and handling raw response
+
+```python
+# Get raw response without automatic decoding
+raw_response = await fga_client.raw_request(
+    operation_name="CustomEndpoint",
+    method="POST",
+    path="/stores/{store_id}/custom-endpoint",
+    path_params={"store_id": FGA_STORE_ID},
+    body={"user": "user:bob", "action": "custom_action"},
+)
+
+# Access the response data
+if raw_response.status == 200:
+    # Manually decode the response
+    result = raw_response.json()
+    if result:
+        print(f"Response: {result}")
+
+    # You can access fields like headers, status code, etc. from raw_response:
+    print(f"Status Code: {raw_response.status}")
+    print(f"Headers: {raw_response.headers}")
+    print(f"Body as text: {raw_response.text()}")
+```
+
+#### Example: Calling a new "Custom Endpoint" endpoint and decoding response into a dictionary
+
+```python
+# Get raw response decoded into a dictionary
+response = await fga_client.raw_request(
+    operation_name="CustomEndpoint",
+    method="POST",
+    path="/stores/{store_id}/custom-endpoint",
+    path_params={"store_id": FGA_STORE_ID},
+    body={"user": "user:bob", "action": "custom_action"},
+)
+
+# The response body is automatically parsed as JSON if possible
+result = response.json()  # Returns dict or None if not JSON
+
+if result:
+    print(f"Response: {result}")
+    # Access fields from the decoded response
+    if "allowed" in result:
+        print(f"Allowed: {result['allowed']}")
+
+print(f"Status Code: {response.status}")
+print(f"Headers: {response.headers}")
+```
+
+#### Example: Calling an existing endpoint with GET
+
+```python
+# Get a list of stores with query parameters
+response = await fga_client.raw_request(
+    operation_name="ListStores",  # Required: descriptive name for the operation
+    method="GET",
+    path="/stores",
+    query_params={
+        "page_size": 10,
+        "continuation_token": "eyJwayI6...",
+    },
+)
+
+stores = response.json()
+print("Stores:", stores)
+```
+
+#### Example: Using Path Parameters
+
+Path parameters are specified in the path using `{param_name}` syntax and are replaced with URL-encoded values from the `path_params` dictionary. If `{store_id}` is present in the path and not provided in `path_params`, it will be automatically replaced with the configured store_id:
+
+```python
+# Using explicit path parameters
+response = await fga_client.raw_request(
+    operation_name="ReadAuthorizationModel",  # Required: descriptive name for the operation
+    method="GET",
+    path="/stores/{store_id}/authorization-models/{model_id}",
+    path_params={
+        "store_id": "your-store-id",
+        "model_id": "your-model-id",
+    },
+)
+
+# Using automatic store_id substitution
+response = await fga_client.raw_request(
+    operation_name="ReadAuthorizationModel",  # Required: descriptive name for the operation
+    method="GET",
+    path="/stores/{store_id}/authorization-models/{model_id}",
+    path_params={
+        "model_id": "your-model-id",
+    },
+)
+```
 
 ### Retries
 

openfga_sdk/client/client.py

@@ -1,6 +1,10 @@
 import asyncio
+import json
+import urllib.parse
 import uuid
 
+from typing import Any
+
 from openfga_sdk.api.open_fga_api import OpenFgaApi
 from openfga_sdk.api_client import ApiClient
 from openfga_sdk.client.configuration import ClientConfiguration
@@ -25,6 +29,7 @@
 from openfga_sdk.client.models.list_objects_request import ClientListObjectsRequest
 from openfga_sdk.client.models.list_relations_request import ClientListRelationsRequest
 from openfga_sdk.client.models.list_users_request import ClientListUsersRequest
+from openfga_sdk.client.models.raw_response import RawResponse
 from openfga_sdk.client.models.read_changes_request import ClientReadChangesRequest
 from openfga_sdk.client.models.tuple import ClientTuple, convert_tuple_keys
 from openfga_sdk.client.models.write_request import ClientWriteRequest
@@ -67,6 +72,10 @@
     WriteAuthorizationModelRequest,
 )
 from openfga_sdk.models.write_request import WriteRequest
+from openfga_sdk.rest import RESTResponse
+from openfga_sdk.telemetry.attributes import (
+    TelemetryAttributes,
+)
 from openfga_sdk.validation import is_well_formed_ulid_string
 
 
@@ -1096,3 +1105,167 @@ def map_to_assertion(client_assertion: ClientAssertion):
             authorization_model_id, api_request_body, **kwargs
         )
         return api_response
+
+    #######################
+    # Raw Request
+    #######################
+    async def raw_request(
+        self,
+        method: str,
+        path: str,
+        query_params: dict[str, str | int | list[str | int]] | None = None,
+        path_params: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
+        body: dict[str, Any] | list[Any] | str | bytes | None = None,
+        operation_name: str | None = None,
+        options: dict[str, int | str | dict[str, int | str]] | None = None,
+    ) -> RawResponse:
+        """
+        Make a raw HTTP request to any OpenFGA API endpoint.
+
+        :param method: HTTP method (GET, POST, PUT, DELETE, PATCH, etc.)
+        :param path: API endpoint path (e.g., "/stores/{store_id}/check" or "/stores")
+        :param query_params: Optional query parameters as a dictionary
+        :param path_params: Optional path parameters to replace placeholders in path
+            (e.g., {"store_id": "abc", "model_id": "xyz"})
+        :param headers: Optional request headers (will be merged with default headers)
+        :param body: Optional request body (dict/list will be JSON serialized, str/bytes sent as-is)
+        :param operation_name: Required operation name for telemetry/logging (e.g., "Check", "Write", "CustomEndpoint")
+        :param options: Optional request options:
+            - headers: Additional headers (merged with headers parameter)
+            - retry_params: Override retry parameters for this request
+            - authorization_model_id: Not used in raw_request, but kept for consistency
+        :return: RawResponse object with status, headers, and body
+        :raises FgaValidationException: If path contains {store_id} but store_id is not configured
+        :raises ApiException: For HTTP errors (with SDK error handling applied)
+        """
+
+        request_headers = dict(headers) if headers else {}
+        if options and options.get("headers"):
+            if isinstance(options["headers"], dict):
+                request_headers.update(options["headers"])
+
+        if not operation_name:
+            raise FgaValidationException("operation_name is required for raw_request")
+
+        resource_path = path
+        path_params_dict = dict(path_params) if path_params else {}
+
+        if "{store_id}" in resource_path and "store_id" not in path_params_dict:
+            store_id = self.get_store_id()
+            if store_id is None or store_id == "":
+                raise FgaValidationException(
+                    "Path contains {store_id} but store_id is not configured. "
+                    "Set store_id in ClientConfiguration, use set_store_id(), or provide it in path_params."
+                )
+            path_params_dict["store_id"] = store_id
+
+        for param_name, param_value in path_params_dict.items():
+            placeholder = f"{{{param_name}}}"
+            if placeholder in resource_path:
+                encoded_value = urllib.parse.quote(str(param_value), safe="")
+                resource_path = resource_path.replace(placeholder, encoded_value)
+        if "{" in resource_path or "}" in resource_path:
+            raise FgaValidationException(
+                f"Not all path parameters were provided for path: {path}"
+            )
+
+        query_params_list = []
+        if query_params:
+            for key, value in query_params.items():
+                if value is None:
+                    continue
+                if isinstance(value, list):
+                    for item in value:
+                        if item is not None:
+                            query_params_list.append((key, str(item)))
+                    continue
+                query_params_list.append((key, str(value)))
+
+        body_params = body
+        if "Content-Type" not in request_headers:
+            if isinstance(body, (dict, list)) or body is None:
+                request_headers["Content-Type"] = "application/json"
+            elif isinstance(body, str):
+                request_headers["Content-Type"] = "text/plain"
+            elif isinstance(body, bytes):
+                request_headers["Content-Type"] = "application/octet-stream"
+            else:
+                request_headers["Content-Type"] = "application/json"
+
+        retry_params = None
+        if options and options.get("retry_params"):
+            retry_params = options["retry_params"]
+        if "Accept" not in request_headers:
+            request_headers["Accept"] = "application/json"
+
+        auth_headers = dict(request_headers) if request_headers else {}
+        await self._api_client.update_params_for_auth(
+            auth_headers,
+            query_params_list,
+            auth_settings=[],
+            oauth2_client=self._api._oauth2_client,
+        )
+
+        telemetry_attributes = None
+        if operation_name:
+            telemetry_attributes = {
+                TelemetryAttributes.fga_client_request_method: operation_name.lower(),
+            }
+            if self.get_store_id():
+                telemetry_attributes[
+                    TelemetryAttributes.fga_client_request_store_id
+                ] = self.get_store_id()
+
+        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=auth_headers if auth_headers else None,
+            body=body_params,
+            response_types_map={},
+            auth_settings=[],
+            _return_http_data_only=True,
+            _preload_content=True,
+            _retry_params=retry_params,
+            _oauth2_client=self._api._oauth2_client,
+            _telemetry_attributes=telemetry_attributes,
+        )
+        rest_response: RESTResponse | None = getattr(
+            self._api_client, "last_response", None
+        )
+
+        if rest_response is None:
+            operation_suffix = (
+                f" (operation: {operation_name})" if operation_name else ""
+            )
+            raise RuntimeError(
+                f"Failed to get response from API client for {method.upper()} "
+                f"request to '{resource_path}'{operation_suffix}. "
+                "This may indicate an internal SDK error, network problem, or client configuration issue."
+            )
+
+        response_body: bytes | str | dict[str, Any] | None = None
+        if rest_response.data is not None:
+            if isinstance(rest_response.data, str):
+                try:
+                    response_body = json.loads(rest_response.data)
+                except (json.JSONDecodeError, ValueError):
+                    response_body = rest_response.data
+            elif isinstance(rest_response.data, bytes):
+                try:
+                    decoded = rest_response.data.decode("utf-8")
+                    try:
+                        response_body = json.loads(decoded)
+                    except (json.JSONDecodeError, ValueError):
+                        response_body = decoded
+                except UnicodeDecodeError:
+                    response_body = rest_response.data
+            else:
+                response_body = rest_response.data
+
+        return RawResponse(
+            status=rest_response.status,
+            headers=dict(rest_response.getheaders()),
+            body=response_body,
+        )

openfga_sdk/client/models/__init__.py

@@ -12,6 +12,7 @@
 from openfga_sdk.client.models.expand_request import ClientExpandRequest
 from openfga_sdk.client.models.list_objects_request import ClientListObjectsRequest
 from openfga_sdk.client.models.list_relations_request import ClientListRelationsRequest
+from openfga_sdk.client.models.raw_response import RawResponse
 from openfga_sdk.client.models.read_changes_request import ClientReadChangesRequest
 from openfga_sdk.client.models.tuple import ClientTuple
 from openfga_sdk.client.models.write_conflict_opts import (
@@ -45,4 +46,5 @@
     "ClientWriteRequestOnMissingDeletes",
     "ConflictOptions",
     "ClientWriteOptions",
+    "RawResponse",
 ]