FastAPI Authentication

pip install unkey.py fastapi uvicorn

from fastapi import HTTPException, Securityfrom fastapi.security import APIKeyHeaderfrom unkey.py import Unkeyimport os# Configure API key headerapi_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)# Initialize Unkey clientunkey = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])async def verify_api_key(api_key: str = Security(api_key_header)):    """Dependency that verifies API keys and returns the verification result."""    if not api_key:        raise HTTPException(            status_code=401,            detail="Missing API key",            headers={"WWW-Authenticate": "ApiKey"},        )    try:        result = await unkey.keys.verify_key_async(            key=api_key,        )    except Exception as e:        raise HTTPException(            status_code=503,            detail="Authentication service unavailable"        )    if not result.valid:        status = 429 if result.code == "RATE_LIMITED" else 401        raise HTTPException(status_code=status, detail=result.code)    return result

from fastapi import FastAPI, Dependsfrom dependencies.auth import verify_api_keyapp = FastAPI()@app.get("/api/data")async def get_data(auth = Depends(verify_api_key)):    return {        "message": "Access granted",        "user": auth.identity.external_id if auth.identity else None,        "remaining_credits": auth.credits,        "metadata": auth.meta,    }@app.get("/api/users")async def get_users(auth = Depends(verify_api_key)):    # Use auth.identity.external_id to scope data access    return {"users": []}

from pydantic import BaseModel, ConfigDictfrom typing import Optional, Anyclass Identity(BaseModel):    id: str    external_id: str    meta: Optional[dict[str, Any]] = Noneclass AuthResult(BaseModel):    valid: bool    code: str    key_id: str    name: Optional[str] = None    identity: Optional[Identity] = None    meta: Optional[dict[str, Any]] = None    credits: Optional[int] = None    expires: Optional[int] = None    enabled: bool = True    permissions: Optional[list[str]] = None    roles: Optional[list[str]] = None    model_config = ConfigDict(extra="allow")  # Allow additional fields from Unkey

from models.auth import AuthResultasync def verify_api_key(api_key: str = Security(api_key_header)) -> AuthResult:    if not api_key:        raise HTTPException(status_code=401, detail="Missing API key")    try:        result = await unkey.keys.verify_key_async(key=api_key)    except Exception as e:        raise HTTPException(status_code=503, detail="Unkey service unavailable")    if not result.valid:        raise HTTPException(status_code=401, detail=result.code)    return AuthResult(        valid=result.valid,        code=result.code,        key_id=result.key_id,        identity=result.identity,        meta=result.meta,        credits=result.credits,    )

from fastapi import HTTPException, Securityfrom fastapi.security import APIKeyHeaderfrom unkey.py import Unkeyimport osapi_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)unkey = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])def require_permission(permission: str):    """Factory that creates a dependency requiring a specific permission."""    async def verify(api_key: str = Security(api_key_header)):        if not api_key:            raise HTTPException(status_code=401, detail="Missing API key")        # Include permission check in verification        result = await unkey.keys.verify_key_async(            key=api_key,            permissions=permission,        )        if not result.valid:            if result.code == "INSUFFICIENT_PERMISSIONS":                raise HTTPException(                    status_code=403,                    detail=f"Permission required: {permission}"                )            raise HTTPException(status_code=401, detail=result.code)        return result    return verify# Pre-built permission checkersrequire_read = require_permission("data.read")require_write = require_permission("data.write")require_admin = require_permission("admin")

from dependencies.auth import require_read, require_write, require_admin@app.get("/api/data")async def read_data(auth = Depends(require_read)):    return {"data": []}@app.post("/api/data")async def create_data(auth = Depends(require_write)):    return {"created": True}@app.delete("/api/users/{user_id}")async def delete_user(user_id: str, auth = Depends(require_admin)):    return {"deleted": user_id}

from fastapi import Responseasync def verify_api_key(    response: Response,    api_key: str = Security(api_key_header)):    if not api_key:        raise HTTPException(status_code=401, detail="Missing API key")    result = await unkey.keys.verify_key_async(        key=api_key,    )    # Add rate limit headers    if result.ratelimits:        rl = result.ratelimits[0]        response.headers["X-RateLimit-Limit"] = str(rl.limit)        response.headers["X-RateLimit-Remaining"] = str(rl.remaining)        response.headers["X-RateLimit-Reset"] = str(rl.reset)    # Add credits header    if result.credits is not None:        response.headers["X-Credits-Remaining"] = str(result.credits)    if not result.valid:        raise HTTPException(            status_code=429 if result.code == "RATE_LIMITED" else 401,            detail=result.code        )    return result

from contextlib import asynccontextmanagerfrom unkey.py import Unkeyimport os# Global client (initialized on startup)unkey_client: Unkey = None@asynccontextmanagerasync def lifespan(app):    global unkey_client    unkey_client = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])    yield    # Cleanup if neededasync def verify_api_key(api_key: str = Security(api_key_header)):    if not api_key:        raise HTTPException(status_code=401, detail="Missing API key")    # Use async method    result = await unkey_client.keys.verify_key_async(        key=api_key,    )    if not result.valid:        raise HTTPException(status_code=401, detail=result.code)    return result

from fastapi import FastAPIfrom dependencies.auth import lifespanapp = FastAPI(lifespan=lifespan)

from fastapi import FastAPI, Depends, HTTPException, Security, Responsefrom fastapi.security import APIKeyHeaderfrom pydantic import BaseModelfrom unkey.py import Unkeyimport osapp = FastAPI(title="My API")# Auth setupapi_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)unkey = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])# Modelsclass DataResponse(BaseModel):    message: str    user: str | None    remaining_credits: int | None# Dependenciesasync def get_auth(    response: Response,    api_key: str = Security(api_key_header)):    if not api_key:        raise HTTPException(status_code=401, detail="Missing API key")    result = await unkey.keys.verify_key_async(key=api_key)    if result.credits is not None:        response.headers["X-Credits-Remaining"] = str(result.credits)    if not result.valid:        raise HTTPException(status_code=401, detail=result.code)    return result# Routes@app.get("/api/data", response_model=DataResponse)async def get_data(auth = Depends(get_auth)):    return DataResponse(        message="Access granted",        user=auth.identity.external_id if auth.identity else None,        remaining_credits=auth.credits,    )@app.get("/health")async def health():    return {"status": "ok"}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="0.0.0.0", port=8000)

UNKEY_ROOT_KEY=... uvicorn main:app --reload