Quickstart: Python and FastAPI

mkdir notes-api && cd notes-api
python3 -m venv .venv && . .venv/bin/activate
pip install "fastapi[standard]" sovaria-sdk
export SOVARIA_URL=https://<project-ref>.sovarias.in SOVARIA_ANON_KEY=<anon-key>

import os

from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel
from sovaria import SovariaError, create_client

URL, ANON_KEY = os.environ["SOVARIA_URL"], os.environ["SOVARIA_ANON_KEY"]
app = FastAPI()


class Credentials(BaseModel):
    email: str
    password: str


class NoteIn(BaseModel):
    body: str


def to_objects(result: dict) -> list[dict]:
    """sql_query answers columns, rows and row_count; rows are lists in column order."""
    return [dict(zip(result["columns"], row)) for row in result["rows"]]


async def as_user(authorization: str = Header(default="")):
    """A client that runs every call as the user whose token the request carries."""
    token = authorization.removeprefix("Bearer ").strip()
    if not token:
        raise HTTPException(401, "Sign in first")
    async with create_client(URL, ANON_KEY) as client:
        client.set_access_token(token)
        yield client


def fail(err: SovariaError):
    raise HTTPException(err.status_code or 502, str(err.detail))


def tokens(t) -> dict:
    """What the app keeps: the access and refresh tokens (no other fields of the sign-in answer)."""
    return t.model_dump(include={"access_token", "refresh_token", "expires_in", "user_id"})


@app.post("/signup")
async def signup(c: Credentials):
    async with create_client(URL, ANON_KEY) as client:
        try:
            return tokens(await client.sign_up(c.email, c.password))
        except SovariaError as err:
            fail(err)


@app.post("/signin")
async def signin(c: Credentials):
    async with create_client(URL, ANON_KEY) as client:
        try:
            return tokens(await client.sign_in(c.email, c.password))
        except SovariaError as err:
            fail(err)


@app.get("/notes")
async def list_notes(sovaria=Depends(as_user)):
    try:
        return to_objects(await sovaria.sql_query("select id, body, created_at from notes order by id desc"))
    except SovariaError as err:
        fail(err)


@app.post("/notes", status_code=201)
async def add_note(note: NoteIn, sovaria=Depends(as_user)):
    try:
        result = await sovaria.sql_query("insert into notes (body) values ($1) returning id, body", params=[note.body])
        return to_objects(result)[0]
    except SovariaError as err:
        fail(err)

fastapi dev main.py   # http://127.0.0.1:8000/docs

TOKEN=$(curl -s 127.0.0.1:8000/signup -H 'Content-Type: application/json' \
  -d '{"email":"ada@example.com","password":"Correct-Horse-9!"}' | jq -r .access_token)
curl -s 127.0.0.1:8000/notes -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' -d '{"body":"From FastAPI"}'
curl -s 127.0.0.1:8000/notes -H "Authorization: Bearer $TOKEN"