dataflow.auth.dependencies

Authentication dependencies for FastAPI routes.

This module provides FastAPI dependency functions for authentication in the Haive API. It includes dependencies for both optional and required authentication, supporting different levels of access control for API endpoints.

The authentication system uses JWT tokens provided through the Authorization header, which are validated against the Supabase authentication service.

Typical usage example:

from fastapi import APIRouter, Depends from haive.dataflow.auth.dependencies import require_auth, get_current_user

router = APIRouter()

# Endpoint requiring authentication @router.get(“/secure”) async def secure_endpoint(user_id: str = Depends(require_auth)):

return {“message”: f”Hello, {user_id}!”}

# Endpoint with optional authentication @router.get(“/public”) async def public_endpoint(user_id: Optional[str] = Depends(get_current_user)):

if user_id:

return {“message”: f”Hello, {user_id}!”}

return {“message”: “Hello, anonymous user!”}

Functions

get_auth_instance()	Get the Supabase authentication instance for dependency injection.
get_current_user([credentials, auth])	Verify the token and return the user ID if valid (optional.
require_auth([user_id])	Require authentication for a route (required authentication).

Module Contents

dataflow.auth.dependencies.get_auth_instance()

Get the Supabase authentication instance for dependency injection.

This function creates and returns a SupabaseAuth instance configured with the server settings from environment variables. It’s used as a FastAPI dependency to provide the authentication service to route handlers.

Returns:

An initialized authentication service instance

Return type:

SupabaseAuth

Examples

>>> from fastapi import Depends
>>> from haive.dataflow.auth.dependencies import get_auth_instance
>>>
>>> async def custom_auth(auth = Depends(get_auth_instance)):
...     # Use auth instance for custom authentication logic
...     return auth.validate_token(token)

async dataflow.auth.dependencies.get_current_user(credentials=Depends(security), auth=Depends(get_auth_instance))

Verify the token and return the user ID if valid (optional. authentication).

This dependency function provides optional authentication for routes. It extracts the JWT token from the Authorization header if present, validates it with Supabase, and returns the user ID if valid. If no token is provided or the token is invalid, it returns None instead of raising an exception.

Parameters:

• credentials (fastapi.security.HTTPAuthorizationCredentials | None) – HTTP Bearer token credentials from the Authorization header

• auth (haive.dataflow.auth.auth.supabase.SupabaseAuth) – Supabase authentication service instance

Returns:

The authenticated user ID if valid, None otherwise

Return type:

Optional[str]

Examples

>>> @router.get("/profile")
>>> async def get_profile(user_id: Optional[str] = Depends(get_current_user)):
...     if user_id:
...         return {"user_id": user_id, "premium": True}
...     else:
...         return {"premium": False}

async dataflow.auth.dependencies.require_auth(user_id=Depends(get_current_user))

Require authentication for a route (required authentication).

This dependency function provides required authentication for routes. It builds on the optional authentication dependency but raises an HTTP exception if no valid authentication is provided, ensuring that the route can only be accessed by authenticated users.

Parameters:

user_id (str | None) – The user ID from the get_current_user dependency

Returns:

The authenticated user ID

Return type:

str

Raises:

HTTPException – 401 Unauthorized if no valid authentication is provided

Examples

>>> @router.post("/secure-endpoint")
>>> async def secure_endpoint(user_id: str = Depends(require_auth)):
...     return {"message": f"Hello, {user_id}!", "status": "authenticated"}