[Python] How to make Open FGA Async Client work with Flask + Gunicorn?

`class FGAClientWrapper: """ A synchronous, gevent-friendly wrapper for the async OpenFGA SDK. """

def __init__(self, call_timeout: float | None = 10.0):
"""
Initializes the configuration, which will be used for each call.
"""
self.config = ClientConfiguration(
api_url=secrets.openfga["FGA_API_URL"],
store_id=secrets.openfga["FGA_STORE_ID"],
authorization_model_id=secrets.openfga["FGA_MODEL_ID"],
credentials=Credentials(
method="api_token", configuration=CredentialConfiguration(api_token=secrets.openfga["FGA_BEARER_TOKEN"])
),
)

self._call_timeout = call_timeout

def _run_sync(self, coro, timeout: float | None = None) -> Any:
try:
# Create a new event loop for each call and run until completion.
# asyncio.wait_for adds the timeout.
timeout_val = timeout if timeout is not None else self._call_timeout
return asyncio.run(asyncio.wait_for(coro, timeout=timeout_val))
except asyncio.TimeoutError as e:
# The asyncio.TimeoutError is the direct cause of the timeout
raise FGAClientError("FGA call timed out") from e
except Exception as e:
# All other exceptions are wrapped
raise FGAClientError("FGA call failed") from e

async def _read_async(self, object_str: str, relation_str: str) -> ReadResponse:
async with OpenFgaClient(self.config) as client:
request = ReadRequestTupleKey(object=object_str, relation=relation_str)
return await client.read(request)

def read(self, object_str: str, relation_str: str) -> ReadResponse:
"""
Reads and returns tuples matching a specific object and relation.
"""
return self._run_sync(self._read_async(object_str, relation_str))`


This is how i am using it right now but creating new loop for each request is not feasible Tried lazy init a background thread and a loop where i queue co routines after taking lock as well but for some api hit it works some the call time out

Any better ways to do this ?

Continue reading...