Skip to main content

FastAPI Module Overview

info

The FastAPI module provides seamless integration between Fluvius domains and FastAPI web applications.

Introduction​

The FastAPI module enables you to:

  • Expose domains as REST APIs
  • Configure authentication and authorization
  • Set up query endpoints
  • Handle media uploads
  • Integrate with MQTT

Quick Start​

Create FastAPI App​

from fastapi import FastAPI
from fluvius.fastapi import create_app, configure_domain_manager

app = create_app()

# Configure domain manager
configure_domain_manager(app, domains=[UserDomain, BankDomain])

Domain Endpoints​

Domains are automatically exposed as REST endpoints:

# POST /api/user/commands/create-user
# GET /api/user/queries/{resource}/{id}
# POST /api/user/queries/{resource}/find

Authentication​

Configure authentication:

from fluvius.fastapi import configure_authentication

configure_authentication(app, backend='keycloak')

Protected Endpoints​

Use the auth_required decorator:

from fluvius.fastapi import auth_required

@app.get('/protected')
@auth_required
async def protected_route():
    return {'message': 'Authenticated'}

Query Endpoints​

Configure query manager for flexible querying:

from fluvius.fastapi import configure_query_manager

configure_query_manager(app, resources=[
    QueryResource('user', UserModel),
    QueryResource('account', AccountModel)
])

Media Handling​

Configure media uploads:

from fluvius.fastapi import configure_media

configure_media(app, filesystem='local', storage_path='/uploads')

MQTT Integration​

Configure MQTT for real-time messaging:

from fluvius.fastapi import configure_mqtt

configure_mqtt(app, broker_url='mqtt://localhost:1883')

Example Application​

from fastapi import FastAPI
from fluvius.fastapi import (
    create_app,
    configure_domain_manager,
    configure_query_manager,
    configure_authentication
)
from user_domain import UserDomain

app = create_app()

# Configure domains
configure_domain_manager(app, domains=[UserDomain])

# Configure queries
configure_query_manager(app, resources=[
    QueryResource('user', UserModel)
])

# Configure auth
configure_authentication(app, backend='keycloak')

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

Next Steps​