Install
Copy
Ask AI
pip install unkey.py fastapi uvicorn
Basic Dependency
dependencies/auth.py
Copy
Ask AI
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
Copy
Ask AI
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
Copy
Ask AI
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
Copy
Ask AI
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
Copy
Ask AI
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")
main.py
Copy
Ask AI
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
Copy
Ask AI
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
Copy
Ask AI
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
Copy
Ask AI
from fastapi import FastAPI
from dependencies.auth import lifespan
app = FastAPI(lifespan=lifespan)
Full Example
main.py
Copy
Ask AI
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)
Copy
Ask AI
UNKEY_ROOT_KEY=... uvicorn main:app --reload

