Scopes

This libray integrates the concept of scopes to provide fine-grained access control for API keys. Scopes are strings that define the permissions associated with an API key. When creating or updating an API key, you can specify the scopes that the key should have. If you define 2 scopes "read" and "write" to an route, an API key must have both scopes to access that route.

Example

Simple

This is the canonical example from examples/example_scopes.py:

Always set a pepper

The default pepper is a placeholder. Set API_KEY_PEPPER (or pass it explicitly to the hashers) in every environment.

import os
from pathlib import Path

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from fastapi_api_key import ApiKey, ApiKeyService
from fastapi_api_key.domain.errors import InvalidScopes
from fastapi_api_key.hasher.argon2 import Argon2ApiKeyHasher
from fastapi_api_key.repositories.sql import SqlAlchemyApiKeyRepository

# Set env var to override default pepper
# Using a strong, unique pepper is crucial for security
# Default pepper is insecure and should not be used in production
pepper = os.getenv("API_KEY_PEPPER")
hasher = Argon2ApiKeyHasher(pepper=pepper)

path = Path(__file__).parent / "db.sqlite3"
database_url = os.environ.get("DATABASE_URL", f"sqlite+aiosqlite:///{path}")

async_engine = create_async_engine(database_url, future=True)
async_session_maker = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def main():
    async with async_session_maker() as async_session:
        repo = SqlAlchemyApiKeyRepository(async_session)

        # Necessary if you don't use your own DeclarativeBase
        await repo.ensure_table()

        svc = ApiKeyService(repo=repo, hasher=hasher)

        # Create an API key without scopes
        bad_entity = ApiKey(name="no-scope-key", scopes=["read"])
        good_entity = ApiKey(name="with-scope-key", scopes=["write"])

        _, bad_api_key = await svc.create(bad_entity)
        _, good_api_key = await svc.create(good_entity)

        print(f"Bad API Key (no required scopes): '{bad_api_key}'")
        print(f"Good API Key (with required scopes): '{good_api_key}'")

        await svc.verify_key(good_api_key, required_scopes=["write"])
        print("Successfully verified good API key with required scopes.")

        try:
            await svc.verify_key(bad_api_key, required_scopes=["write"])
        except InvalidScopes as e:
            print(f"Failed to verify bad API key with required scopes: {e}")


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

FastAPI

You can create security Depends with required scopes like this:

import os
from pathlib import Path
from typing import AsyncIterator

from fastapi import FastAPI, Depends, APIRouter
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from fastapi_api_key import ApiKey, ApiKeyService
from fastapi_api_key.hasher.argon2 import Argon2ApiKeyHasher
from fastapi_api_key.repositories.sql import SqlAlchemyApiKeyRepository
from fastapi_api_key.api import create_api_keys_router, create_depends_api_key

# Set env var to override default pepper
# Using a strong, unique pepper is crucial for security
# Default pepper is insecure and should not be used in production
pepper = os.getenv("API_KEY_PEPPER")
hasher = Argon2ApiKeyHasher(pepper=pepper)

path = Path(__file__).parent / "db.sqlite3"
database_url = os.environ.get("DATABASE_URL", f"sqlite+aiosqlite:///{path}")

async_engine = create_async_engine(database_url, future=True)
async_session_maker = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

app = FastAPI(title="API with API Key Management")


async def inject_async_session() -> AsyncIterator[AsyncSession]:
    """Dependency to provide an active SQLAlchemy async session."""
    async with async_session_maker() as session:
        async with session.begin():
            yield session


async def inject_svc_api_keys(async_session: AsyncSession = Depends(inject_async_session)) -> ApiKeyService:
    """Dependency to inject the API key service with an active SQLAlchemy async session."""
    repo = SqlAlchemyApiKeyRepository(async_session)

    # Necessary if you don't use your own DeclarativeBase
    await repo.ensure_table()

    return ApiKeyService(repo=repo, hasher=hasher)


# Create security dependency that requires "write" scope
security = create_depends_api_key(inject_svc_api_keys, required_scopes=["write"])
router_protected = APIRouter(prefix="/protected", tags=["Protected"])

router = APIRouter(prefix="/api-keys", tags=["API Keys"])
router_api_keys = create_api_keys_router(
    inject_svc_api_keys,
    router=router,
)


@router_protected.get("/")
async def read_protected_data(api_key: ApiKey = Depends(security)):
    return {
        "message": "This is protected data",
        "apiKey": {
            "id": api_key.id_,
            "name": api_key.name,
            "description": api_key.description,
            "isActive": api_key.is_active,
            "createdAt": api_key.created_at,
            "expiresAt": api_key.expires_at,
            "lastUsedAt": api_key.last_used_at,
        },
    }


app.include_router(router_api_keys)
app.include_router(router_protected)


async def main():
    async with async_session_maker() as async_session:
        repo = SqlAlchemyApiKeyRepository(async_session)

        # Necessary if you don't use your own DeclarativeBase
        await repo.ensure_table()

        svc = ApiKeyService(repo=repo, hasher=hasher)

        # Create an API key without scopes
        bad_entity = ApiKey(name="no-scope-key", scopes=["read"])
        good_entity = ApiKey(name="with-scope-key", scopes=["write"])

        _, bad_api_key = await svc.create(bad_entity)
        _, good_api_key = await svc.create(good_entity)

        print(f"Bad API Key (no required scopes): '{bad_api_key}'")
        print(f"Good API Key (with required scopes): '{good_api_key}'")

        await async_session.commit()


if __name__ == "__main__":
    import asyncio

    # Create an invalid and a valid API key for testing
    asyncio.run(main())

    import uvicorn

    # Run the FastAPI app and test the 2 api keys against the protected endpoint
    uvicorn.run(app, host="localhost", port=8000)