Python: install and configure
Install sovaria-sdk, async and sync clients, configuration from arguments, files or the environment, acting as a user on a server, retries, errors, and Flask, Django and FastAPI integration.
Install
pip install sovaria-sdkPython 3.11 or newer. The package is imported as sovaria.
Async and sync clients
from sovaria import create_client
async with create_client("https://<project-ref>.sovarias.in", "<anon-key>") as sovaria:
result = await sovaria.sql_query("select now()")Use the async client in async frameworks (FastAPI, aiohttp, asyncio scripts) and the sync client in Django, Flask and plain scripts; both have the same methods. Open the async client with async with (it holds the HTTP connection pool) and the sync client with with or call close() when done.
Act as the user who called your server
from sovaria import create_client
async with create_client(SOVARIA_URL, SOVARIA_ANON_KEY) as client:
client.set_access_token(access_token, expires_in=900) # every call now runs as that user
result = await client.sql_query("select id, total from orders")Errors and retries
from sovaria import SovariaError
try:
await sovaria.sql_query("select * from orders")
except SovariaError as err:
if err.status_code == 401:
... # sign in again
elif err.status_code == 429:
... # back off
else:
raiseSovariaError carries status_code (0 for a network error) and detail (the server's message). Requests that fail with 429, 500, 502, 503, 504 or a network error are retried — 4 attempts in all, with jittered backoff and at least the Retry-After time. Writes are retried too and applied once: as in JavaScript, every write carries one Idempotency-Key from its first attempt, so a retry after a lost response gets the stored answer instead of running again. To repeat a write safely from your own code, pass a key you can recreate: await sovaria.request("POST", path, json=body, headers={"Idempotency-Key": f"order-{order_id}"}) (Reliability).
FastAPI
from fastapi import Depends, FastAPI, Header, HTTPException
from sovaria import SovariaError, create_client
app = FastAPI()
async def sovaria_as_user(authorization: str = Header(default="")):
token = authorization.removeprefix("Bearer ").strip()
if not token:
raise HTTPException(401, "Sign in")
async with create_client(SOVARIA_URL, SOVARIA_ANON_KEY) as client:
client.set_access_token(token, expires_in=900)
yield client
@app.get("/orders")
async def orders(sovaria=Depends(sovaria_as_user)):
try:
return await sovaria.sql_query("select id, total from orders order by id desc limit 20")
except SovariaError as err:
raise HTTPException(err.status_code or 502, str(err))Django and Flask
from django.http import JsonResponse
from sovaria import create_sync_client
def orders(request):
token = request.headers.get("Authorization", "").removeprefix("Bearer ")
with create_sync_client(SOVARIA_URL, SOVARIA_ANON_KEY) as client:
client.set_access_token(token, expires_in=900)
return JsonResponse(client.sql_query("select id, total from orders"))Every method, with signatures: Python SDK.