FastAPI

El SDK de Python en FastAPI, con las dependencias del framework.

bash
uv add andco-sdk

El SDK de Python es un cliente confidencial de backend: no presenta nada, no abre ventanas y no guarda sesión. Conduce OAuth y llama a la API por cuenta de una persona.

La instancia

app/andco.pypython
import os
from andco_sdk import AndcoClient

andco = AndcoClient(
    client_id=os.environ["ANDCO_CLIENT_ID"],
    client_secret=os.environ["ANDCO_CLIENT_SECRET"],
)

Se construye una vez y se comparte entre peticiones concurrentes. Es seguro porque es un valor: no tiene ningún campo donde el token de una persona pueda quedar y aparecer en la respuesta de otra.

Iniciar la autorización

app/routes/auth.pypython
from fastapi import APIRouter, Request
from fastapi.responses import RedirectResponse

from app.andco import andco

router = APIRouter()


@router.get("/auth/andco")
async def start(request: Request) -> RedirectResponse:
    authorization = await andco.oauth.create_authorization_request(
        redirect_to=f"{settings.app_origin}/auth/andco/callback",
        scopes=["openid", "email", "profile"],
    )
    request.session["andco_tx"] = {
        "authorization_url": authorization.authorization_url,
        "code_verifier": authorization.code_verifier,
        "state": authorization.state,
        "redirect_to": authorization.redirect_to,
    }
    return RedirectResponse(authorization.authorization_url)

El code_verifier es lo que no puede perderse: sin él, el código que vuelve no se puede canjear. Guárdalo en la sesión del navegador, nunca en la URL.

Recibir el callback

python
from andco_sdk import AndcoAuthorizationRequest, AndcoError


@router.get("/auth/andco/callback")
async def callback(request: Request) -> RedirectResponse:
    stored = request.session.pop("andco_tx", None)
    if not stored:
        raise HTTPException(status_code=400, detail="missing_transaction")

    try:
        session = await andco.oauth.exchange_callback(
            callback_url=str(request.url),
            request=AndcoAuthorizationRequest(**stored),
        )
    except AndcoError as error:
        raise HTTPException(status_code=400, detail=error.code) from error

    await save_to_your_session(request, session)
    return RedirectResponse("/")

La sesión como dependencia

Aquí es donde FastAPI se gana su lugar: la credencial ligada llega como dependencia y las rutas no vuelven a mencionarla.

app/deps.pypython
from typing import Annotated

from andco_sdk import BoundAndcoClient
from fastapi import Depends, HTTPException, Request

from app.andco import andco


async def bound_andco(request: Request) -> BoundAndcoClient:
    session = await session_from_cookie(request)
    if session is None:
        raise HTTPException(status_code=401, detail="unauthenticated")
    return andco.with_session(session)


Andco = Annotated[BoundAndcoClient, Depends(bound_andco)]
app/routes/accounts.pypython
@router.get("/accounts")
async def accounts(client: Andco) -> dict:
    return await client.rest.get("accounts")

with_session devuelve un cliente nuevo y no muta la instancia compartida, que es lo que hace segura una instancia de módulo bajo concurrencia.

Acepta tres formas, y ninguna renueva por sí sola — son una consulta directa al token:

python
andco.with_session("a-bare-access-token")
andco.with_session(session)  # a full session — same lookup, still no refresh
andco.with_session(my_credentials)  # your own implementation, can refresh if you write it

Errores

El SDK de Python lanza excepciones en vez de devolver resultados, porque es lo que se espera de una biblioteca de Python.

python
from andco_sdk import AndcoError

try:
    accounts = await client.rest.get("accounts")
except AndcoError as error:
    if error.status_code == 403:
        raise HTTPException(status_code=403, detail="missing_permission") from error
    raise

Renovar

python
refreshed = await andco.oauth.refresh(session.refresh_token)
await save_to_your_session(request, refreshed)

Los refresh tokens rotan: el nuevo reemplaza al anterior y el anterior deja de servir.

Webhooks

app/routes/webhooks.pypython
from andco_sdk import (
    ANDCO_WEBHOOK_EVENT_ID_HEADER,
    ANDCO_WEBHOOK_SIGNATURE_HEADER,
    ANDCO_WEBHOOK_TIMESTAMP_HEADER,
    verify_webhook,
)


@router.post("/webhooks/andco", status_code=204)
async def webhook(request: Request) -> None:
    raw = await request.body()
    trusted = verify_webhook(
        body=raw,
        signature=request.headers.get(ANDCO_WEBHOOK_SIGNATURE_HEADER),
        event_id=request.headers.get(ANDCO_WEBHOOK_EVENT_ID_HEADER),
        timestamp=request.headers.get(ANDCO_WEBHOOK_TIMESTAMP_HEADER),
        secret=settings.andco_webhook_secret,
    )
    if not trusted:
        raise HTTPException(status_code=400, detail="invalid_signature")

    await handle_event(json.loads(raw))

await request.body() y no await request.json(): la firma cubre los bytes crudos, y volver a serializar el objeto produce otros bytes.