FastAPI authentication

import os
from typing import Annotated

import httpx
from fastapi import Depends, HTTPException, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from unkey.py import Unkey, errors, models

unkey = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])
bearer = HTTPBearer(auto_error=True)

STATUS = {
    "RATE_LIMITED": 429,
    "USAGE_EXCEEDED": 402,
    "FORBIDDEN": 403,
    "INSUFFICIENT_PERMISSIONS": 403,
}


def require_key(permissions: str | None = None):
    def dependency(
        response: Response,
        credentials: Annotated[HTTPAuthorizationCredentials, Depends(bearer)],
    ) -> models.V2KeysVerifyKeyResponseData:
        try:
            res = unkey.keys.verify_key(key=credentials.credentials, permissions=permissions)
        except (errors.UnkeyError, httpx.HTTPError) as e:
            raise HTTPException(status_code=503, detail="authentication unavailable") from e

        data = res.data
        if data.ratelimits:
            # Report the tightest limit that was checked.
            rl = min(data.ratelimits, key=lambda r: r.remaining)
            response.headers["X-RateLimit-Limit"] = str(rl.limit)
            response.headers["X-RateLimit-Remaining"] = str(rl.remaining)
            # rl.reset is a Unix timestamp in milliseconds; the header is conventionally seconds.
            response.headers["X-RateLimit-Reset"] = str(rl.reset // 1000)

        if not data.valid:
            code = data.code
            raise HTTPException(status_code=STATUS.get(code, 401), detail=code)
        return data

    return dependency

from typing import Annotated

from fastapi import Depends, FastAPI
from unkey.py import models

from .auth import require_key

app = FastAPI()


@app.get("/items")
def list_items(verification: Annotated[models.V2KeysVerifyKeyResponseData, Depends(require_key())]):
    owner = verification.identity.external_id if verification.identity else None
    return {"items": [], "owner": owner, "credits_left": verification.credits}


@app.get("/reports")
def reports(verification: Annotated[models.V2KeysVerifyKeyResponseData, Depends(require_key("reports.read"))]):
    return {"reports": [], "key": verification.key_id}