Skip to main content

Authentication

info

Fluvius FastAPI module provides flexible authentication backends including Keycloak integration.

Overview​

The FastAPI module supports multiple authentication backends:

  • Keycloak: OAuth2/OIDC authentication
  • Custom: Implement your own authentication backend
  • Session-based: Cookie-based sessions
  • Token-based: JWT token authentication

Configure Authentication​

Keycloak Integration​

from fluvius.fastapi import configure_authentication

configure_authentication(
    app,
    backend='keycloak',
    keycloak_url='https://keycloak.example.com',
    realm='your-realm',
    client_id='your-client-id',
    client_secret='your-client-secret'
)

Custom Backend​

from fluvius.fastapi import configure_authentication

def custom_auth_handler(request):
    # Your authentication logic
    token = request.headers.get('Authorization')
    # Validate token and return user info
    return {'user_id': '...', 'realm': '...'}

configure_authentication(
    app,
    backend='custom',
    handler=custom_auth_handler
)

Protected Endpoints​

Use the auth_required decorator:

from fluvius.fastapi import auth_required

@app.get('/api/protected')
@auth_required
async def protected_route(request):
    user_id = request['user_id']
    realm = request['realm']
    return {'message': f'Hello {user_id}'}

Authorization Context​

Authentication information is available in the request context:

@app.post('/api/user/commands/create-user')
@auth_required
async def create_user(request, payload):
    # Access auth context
    user_id = request['user_id']
    realm = request['realm']
    organization_id = request.get('organization_id')
    
    # Use in domain context
    ctx = SanicContext.create(
        namespace='app-user',
        user_id=user_id,
        realm=realm
    )
    
    domain = UserDomain(ctx)
    # ... process command

Domain Integration​

Domains automatically receive authentication context:

# Context is automatically passed to domains
with domain.session(None, **request_context):
    command = domain.create_command('create-user', payload)
    response = await domain.process_command(command)

Policy Enforcement​

Combine with Casbin for policy-based authorization:

from fluvius.casbin import PolicyManager

# Policies are checked automatically when processing commands
# if domain has __policymgr__ configured

Next Steps​