Guías
Ver como MarkdownFastAPI
El SDK de Python en FastAPI, con las dependencias del framework.
uv add andco-sdkEl 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
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
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
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.
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)]@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:
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 itErrores
El SDK de Python lanza excepciones en vez de devolver resultados, porque es lo que se espera de una biblioteca de 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
raiseRenovar
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
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.
