Skip to main content

Add API Endpoint

info

Learn how to add API endpoints in Fluvius Framework with FastAPI integration.

Automatic Endpoint Generation​

Fluvius automatically generates endpoints from domains:

from fluvius.fastapi import FluviusFastAPI
from myapp.domains.user import UserDomain

app = FluviusFastAPI()
app.register_domain(UserDomain)

# Endpoints automatically created:
# POST /api/user/create-user
# POST /api/user/update-user
# POST /api/user/deactivate-user
# GET /api/user/{user_id}

Custom Endpoints​

Basic Endpoint​

from fastapi import APIRouter
from fluvius.fastapi import FluviusFastAPI

app = FluviusFastAPI()

@app.get("/api/users")
async def list_users():
    # Custom endpoint logic
    users = await domain.statemgr.find('user')
    return users

Endpoint with Path Parameters​

@app.get("/api/users/{user_id}")
async def get_user(user_id: str):
    user = await domain.statemgr.fetch('user', user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Endpoint with Request Body​

from pydantic import BaseModel

class CreateUserRequest(BaseModel):
    name: str
    email: str

@app.post("/api/users")
async def create_user(request: CreateUserRequest):
    command = domain.create_command('create-user', {
        'name': request.name,
        'email': request.email
    })
    response = await domain.process_command(command)
    return response

Endpoint Groups​

Router Groups​

from fastapi import APIRouter

router = APIRouter(prefix="/api/v1", tags=["users"])

@router.get("/users")
async def list_users():
    pass

@router.post("/users")
async def create_user():
    pass

app.include_router(router)

Authentication​

Protected Endpoints​

from fluvius.fastapi.authentication import require_auth

@app.get("/api/users")
@require_auth
async def list_users():
    # Requires authentication
    pass

Role-Based Endpoints​

from fluvius.fastapi.authentication import require_role

@app.delete("/api/users/{user_id}")
@require_role("admin")
async def delete_user(user_id: str):
    # Requires admin role
    pass

Request Validation​

Pydantic Models​

from pydantic import BaseModel, EmailStr, Field

class CreateUserRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

@app.post("/api/users")
async def create_user(request: CreateUserRequest):
    # Request is automatically validated
    pass

Response Models​

Response Types​

from pydantic import BaseModel

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    active: bool

@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: str):
    user = await domain.statemgr.fetch('user', user_id)
    return user

Error Handling​

Custom Error Handlers​

from fastapi import HTTPException

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"error": str(exc)}
    )

@app.post("/api/users")
async def create_user(request: CreateUserRequest):
    try:
        response = await domain.process_command(command)
        return response
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

Next Steps​