pip install avnology-id fastapi uvicorn

# lib/avnology.py
from avnology_id import AsyncAvnologyId

client = AsyncAvnologyId(
    base_url="https://api.id.avnology.com",
    client_id="app_abc123",
    client_secret="sk_live_...",
)

from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from lib.avnology import client

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    result = await client.oauth.introspect_token(token=credentials.credentials)

    if not result.active:
        raise HTTPException(status_code=401, detail="Invalid or expired token")

    return {
        "user_id": result.sub,
        "scopes": result.scope.split(),
        "org_id": result.org_id,
    }

from fastapi import FastAPI

app = FastAPI()

@app.get("/api/profile")
async def get_profile(user: dict = Depends(get_current_user)):
    full_user = await client.admin.get_user(user_id=user["user_id"])
    return {
        "email": full_user.email,
        "name": full_user.name,
    }

def require_permission(relation: str, object_fn):
    async def checker(request: Request, user: dict = Depends(get_current_user)):
        obj = object_fn(request)
        allowed = await client.permissions.check(
            subject=f"user:{user['user_id']}",
            relation=relation,
            object=obj,
        )
        if not allowed:
            raise HTTPException(status_code=403, detail="Permission denied")
        return user
    return checker

# Usage:
@app.put("/api/projects/{project_id}")
async def update_project(
    project_id: str,
    user: dict = Depends(require_permission("editor", lambda r: f"project:{r.path_params['project_id']}")),
):
    return {"updated": True}

from avnology_id.webhook import verify_signature

@app.post("/webhooks/avnology")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Avnology-Signature", "")

    if not verify_signature(body, signature, WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()

    match event["type"]:
        case "user.created":
            await handle_user_created(event["data"])
        case "user.deleted":
            await handle_user_deleted(event["data"])

    return {"received": True}

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: client is already initialized
    yield
    # Shutdown: close the HTTP client
    await client.close()

app = FastAPI(lifespan=lifespan)