sagecoders-events

SDK Python de SageEvents, le bus d'événements de l'écosystème SageCoders. Il ne dépend d'aucun framework : Django, FastAPI, Flask ou script. Il signe exactement comme le SDK JS @sagecoders/events : un test commun le garantit.

pip install sagecoders-events

Toutes les erreurs sont des SageEventsError avec un code stable et un status HTTP.

Publier

from sagecoders_auth import ServiceAccount
from sagecoders_events import SageEvents

svc = ServiceAccount(issuer=SAGEAUTH_ISSUER, key=SAGEAUTH_SERVICE_KEY_B64, sageauth_project_id=SAGEAUTH_ADMIN_PROJECT_ID)
bus = SageEvents(SAGE_EVENTS_URL, token=lambda: svc.get_token(SAGE_EVENTS_PROJECT_ID), app="sage-hotel")

bus.publish(
    "sagehotel.booking.created",
    {"reservation_id": "R-1", "guest_phone": "+2250700000000"},
    org_id="org_ivoire",
    subject_id="R-1",
    sensitive=["data.guest_phone"],   # masqué dans la console sans permission
)

Le type est au format produit.objet.action, dans l'espace de noms de votre projet SageAuth. SageEvents vérifie le droit de publier auprès de SageAuth.

Pour publier en réaction à un événement reçu, reprenez son correlation_id et passez son id en causation_id. La console reconstruit ainsi l'arbre complet de l'action.

S'abonner (au déploiement)

r = bus.subscribe("https://api.sagehotel.com/sageevents/", ["sagecoders.user.deleted", "sagestock.*"])
SAGE_EVENTS_SECRET = r["secret"]   # à garder dans le coffre de secrets

Recevoir

Chaque livraison est un POST signé (SageEvents-Signature). Votre fonction renvoie PROCESSED (ou None), ou IGNORED si l'événement ne vous concerne pas. Une exception déclenche une nouvelle tentative (5 s, 15 s, 1 min… puis lettre morte, rejouable depuis la console).

from sagecoders_events import IGNORED, handle

def traiter(event):
    if event["type"] != "sagecoders.user.deleted":
        return IGNORED
    delier_compte(event["data"]["sub"])

Django

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

@csrf_exempt
@require_POST
def sageevents(request):
    r = handle(request.body, request.headers.get("SageEvents-Signature"), settings.SAGE_EVENTS_SECRET, traiter)
    response = HttpResponse(r.body, status=r.status)
    for k, v in r.headers.items():
        response[k] = v
    return response

FastAPI

from fastapi import Request, Response

@app.post("/sageevents/")
async def sageevents(request: Request):
    r = handle(await request.body(), request.headers.get("SageEvents-Signature"), SAGE_EVENTS_SECRET, traiter)
    return Response(r.body, status_code=r.status, headers=r.headers)

Flask : r = handle(request.get_data(), request.headers.get("SageEvents-Signature"), SECRET, traiter), puis return r.body, r.status, r.headers.

Passez toujours le corps brut, avant tout décodage JSON. Répondez en moins de 10 s ; un traitement long va dans une file.

Développement

python -m venv .venv && .venv/Scripts/pip install -e .     # (bin/ sous Linux et macOS)
python -m unittest discover -s tests                       # hors ligne, SageEvents simulé