33 lines
1.0 KiB
Python
33 lines
1.0 KiB
Python
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, Request, Security, status
|
|
from fastapi.security import APIKeyHeader
|
|
|
|
from users import AuthenticatedUser, UserStore
|
|
|
|
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
|
|
|
|
|
|
class ApiKeyAuthenticator:
|
|
def __init__(self, user_store: UserStore) -> None:
|
|
self._user_store = user_store
|
|
|
|
async def authenticate(self, api_key: str | None) -> AuthenticatedUser | None:
|
|
return await self._user_store.authenticate(api_key)
|
|
|
|
|
|
async def require_api_key(
|
|
request: Request,
|
|
api_key: Annotated[str | None, Security(api_key_header)],
|
|
) -> AuthenticatedUser:
|
|
authenticator: ApiKeyAuthenticator = request.app.state.authenticator
|
|
user = await authenticator.authenticate(api_key)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="api_key is required",
|
|
)
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[AuthenticatedUser, Depends(require_api_key)] |