FastAPI Authentication

Implement API key authentication in FastAPI using Unkey with a dependency injection pattern. Protect endpoints with automatic key verification.

A clean, reusable pattern using FastAPI’s dependency injection system.

Install

pip install unkey.py fastapi uvicorn

Basic Dependency

dependencies/auth.py

from fastapi import HTTPException, Security
from fastapi.security import APIKeyHeader
from unkey.py import Unkey
import os

# Configure API key header
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

# Initialize Unkey client
unkey = 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

Use in Routes

main.py

from fastapi import FastAPI, Depends
from dependencies.auth import verify_api_key

app = 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": []}

---

Typed Response Model

Create a Pydantic model for the auth result:

models/auth.py

from pydantic import BaseModel, ConfigDict
from typing import Optional, Any

class Identity(BaseModel):
    id: str
    external_id: str
    meta: Optional[dict[str, Any]] = None

class 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

dependencies/auth.py

from models.auth import AuthResult

async 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,
    )

---

Permission-Based Access

Create dependencies for different permission levels:

dependencies/auth.py

from fastapi import HTTPException, Security
from fastapi.security import APIKeyHeader
from unkey.py import Unkey
import os

api_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 checkers
require_read = require_permission("data.read")
require_write = require_permission("data.write")
require_admin = require_permission("admin")

Use in routes:

main.py

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}

---

Rate Limit Headers

Return rate limit info in response headers:

dependencies/auth.py

from fastapi import Response

async 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

---

Async Client

For better performance with async FastAPI:

dependencies/auth.py

from contextlib import asynccontextmanager
from unkey.py import Unkey
import os

# Global client (initialized on startup)
unkey_client: Unkey = None

@asynccontextmanager
async def lifespan(app):
    global unkey_client
    unkey_client = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])
    yield
    # Cleanup if needed

async 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

main.py

from fastapi import FastAPI
from dependencies.auth import lifespan

app = FastAPI(lifespan=lifespan)

---

Full Example

main.py

from fastapi import FastAPI, Depends, HTTPException, Security, Response
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
from unkey.py import Unkey
import os

app = FastAPI(title="My API")

# Auth setup
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
unkey = Unkey(root_key=os.environ["UNKEY_ROOT_KEY"])

# Models
class DataResponse(BaseModel):
    message: str
    user: str | None
    remaining_credits: int | None

# Dependencies
async 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)

Run with:

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