Django

El SDK de Python sobre las sesiones y las vistas de Django. Sin paquete de integración.

bash
uv add andco-sdk

No existe un paquete de integración de Andco para Django. Lo que ves aquí es el SDK usado directamente sobre los mecanismos del framework: request.session, vistas y middleware. No hay nada que instalar aparte de andco-sdk.

El SDK es asíncrono, así que las vistas que lo usan son async def. Django las soporta desde la versión 4.1 y no requiere ASGI para vistas asíncronas, aunque con ASGI rinden mejor.

La instancia

andco/client.pypython
from andco_sdk import AndcoClient
from django.conf import settings

andco = AndcoClient(
    client_id=settings.ANDCO_CLIENT_ID,
    client_secret=settings.ANDCO_CLIENT_SECRET,
)

Un objeto de módulo, compartido por todas las peticiones. No guarda sesión, así que compartirlo no mezcla credenciales.

Configuración

settings.pypython
ANDCO_CLIENT_ID = os.environ["ANDCO_CLIENT_ID"]
ANDCO_CLIENT_SECRET = os.environ["ANDCO_CLIENT_SECRET"]
ANDCO_WEBHOOK_SECRET = os.environ["ANDCO_WEBHOOK_SECRET"]
APP_ORIGIN = os.environ["APP_ORIGIN"]

Iniciar la autorización

andco/views.pypython
from dataclasses import asdict

from django.http import HttpRequest, HttpResponseRedirect
from django.conf import settings

from .client import andco


async def start(request: HttpRequest) -> HttpResponseRedirect:
    authorization = await andco.oauth.create_authorization_request(
        redirect_to=f"{settings.APP_ORIGIN}/andco/callback",
        scopes=["openid", "email", "profile"],
    )
    request.session["andco_tx"] = asdict(authorization)
    return HttpResponseRedirect(authorization.authorization_url)

request.session de Django ya es el lugar correcto: está ligado a la cookie del navegador, es del lado servidor y el code_verifier nunca viaja al cliente.

Recibir el callback

python
from andco_sdk import AndcoAuthorizationRequest, AndcoError
from django.http import HttpResponseBadRequest


async def callback(request: HttpRequest) -> HttpResponse:
    stored = request.session.pop("andco_tx", None)
    if not stored:
        return HttpResponseBadRequest("missing_transaction")

    try:
        session = await andco.oauth.exchange_callback(
            callback_url=request.build_absolute_uri(),
            request=AndcoAuthorizationRequest(**stored),
        )
    except AndcoError as error:
        return HttpResponseBadRequest(error.code)

    request.session["andco"] = {
        "access_token": session.access_token,
        "refresh_token": session.refresh_token,
        "expires_at": session.expires_at,
        "scopes": list(session.scopes),
    }
    return HttpResponseRedirect("/")

Si tu proyecto usa el modelo de usuario de Django, este es el punto donde creas o recuperas el usuario y llamas a login(). La sesión de Andco y la sesión de Django son cosas distintas: la primera autoriza contra Andco, la segunda identifica en tu aplicación.

andco/urls.pypython
from django.urls import path

from . import views

urlpatterns = [
    path("andco/start", views.start, name="andco-start"),
    path("andco/callback", views.callback, name="andco-callback"),
]

Middleware que liga la credencial

andco/middleware.pypython
from andco_sdk import AndcoSession
from django.utils.deprecation import MiddlewareMixin

from .client import andco


class AndcoMiddleware(MiddlewareMixin):
    """Leaves a client bound to this request's credential in `request.andco`."""

    def process_request(self, request):
        stored = request.session.get("andco")
        request.andco = None
        if stored:
            request.andco = andco.with_session(AndcoSession(**stored, user=None))
settings.pypython
MIDDLEWARE = [
    # …
    "django.contrib.sessions.middleware.SessionMiddleware",
    "andco.middleware.AndcoMiddleware",
]

Tiene que ir después del middleware de sesiones de Django, porque lee request.session.

Usarlo en una vista

python
from django.http import JsonResponse


async def accounts(request):
    if request.andco is None:
        return JsonResponse({"error": "unauthenticated"}, status=401)
    return JsonResponse(await request.andco.rest.get("accounts"))

Webhooks

andco/webhooks.pypython
import json

from andco_sdk import (
    ANDCO_WEBHOOK_EVENT_ID_HEADER,
    ANDCO_WEBHOOK_SIGNATURE_HEADER,
    ANDCO_WEBHOOK_TIMESTAMP_HEADER,
    verify_webhook,
)
from django.conf import settings
from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt


@csrf_exempt
async def webhook(request):
    trusted = verify_webhook(
        body=request.body,
        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:
        return HttpResponseBadRequest("invalid_signature")

    await handle_event(json.loads(request.body))
    return HttpResponse(status=204)

csrf_exempt es correcto aquí: Andco no tiene tu token CSRF, y la firma HMAC es la autenticación de esta ruta. No lo pongas en ninguna otra vista.

request.body son los bytes crudos, que es lo que la firma cubre.

Vistas síncronas

Si tu proyecto no puede usar vistas asíncronas todavía, asgiref envuelve las llamadas:

python
from asgiref.sync import async_to_sync


def accounts(request):
    return JsonResponse(async_to_sync(request.andco.rest.get)("accounts"))

Funciona, pero bloquea el hilo mientras dura la llamada. Es una salida de compromiso, no el camino recomendado.