Skip to main content

ADR-003: FastAPI as Web Framework

Status

Accepted

Context

The original Heka platform was built on Flask with Gunicorn, which led to several limitations:
  • Synchronous by default: Poor performance for I/O heavy operations
  • No native async support: Difficult to implement websockets, SSE
  • Manual API documentation: OpenAPI specs maintained separately from code
  • Limited type safety: No automatic request/response validation
  • Concurrency issues: Problems with worker processes and shared state
For Tasteful, we needed a modern web framework that could address these issues while maintaining developer productivity.

Decision

We selected FastAPI as the foundation web framework for Tasteful.

Key Advantages

  1. Native Async Support: Built on ASGI (Starlette) for true asynchronous processing
  2. Automatic API Documentation: OpenAPI/Swagger docs generated from code
  3. Type Safety: Pydantic integration for request/response validation
  4. Performance: Among the fastest Python web frameworks
  5. Modern Standards: Built-in support for modern web standards (JSON, WebSockets, etc.)

Implementation Pattern

from fastapi import FastAPI
from tasteful.base_flavor import BaseFlavor, BaseController
from tasteful.decorators.controller import Get, Post

class UserController(BaseController):
    def __init__(self, user_service: UserService):
        super().__init__(prefix="/users", tags=["users"])
        self.user_service = user_service
    
    @Get("/{user_id}")
    async def get_user(self, user_id: int) -> UserResponse:
        # Automatic validation of user_id as int
        # Automatic serialization of UserResponse
        return await self.user_service.get_user(user_id)
    
    @Post("/")
    async def create_user(self, user: UserCreate) -> UserResponse:
        # Automatic validation of request body against UserCreate schema
        return await self.user_service.create_user(user)

class UserFlavor(BaseFlavor):
    def __init__(self):
        super().__init__(
            controller=UserController,
            services=[UserService],
            repositories=[UserRepository],
            config=UserConfig
        )

Integration with Tasteful Architecture

class TastefulApp:
    def __init__(self, title: str, version: str, flavors: List[BaseFlavor]):
        # FastAPI as the core application
        self.app = FastAPI(title=title, version=version)
        
        # Register flavor controllers as FastAPI routers
        for flavor_instance in flavors:
            self.app.include_router(flavor_instance.controller.router)

Consequences

Positive

  • Async Performance: Better handling of I/O operations and concurrent requests
  • Developer Experience: Automatic API docs, type hints, IDE support
  • Validation: Built-in request/response validation reduces bugs
  • Standards Compliance: OpenAPI 3.0, JSON Schema support out of the box
  • Future-Proof: Modern async architecture supports websockets, SSE, etc.
  • Ecosystem: Large ecosystem of FastAPI extensions and middleware

Negative

  • Learning Curve: Team needs to understand async/await patterns
  • Complexity: More complex than synchronous frameworks for simple use cases
  • Debugging: Async stack traces can be more difficult to debug
  • Dependencies: Heavier dependency footprint than minimal frameworks

Trade-offs

  • Performance vs. Simplicity: Better performance but requires async understanding
  • Type Safety vs. Flexibility: More rigid but catches errors earlier
  • Auto-documentation vs. Control: Generated docs are convenient but less customizable

Alternatives Considered

  1. Flask + Flask-RESTX: Continue with existing stack
    • Rejected: Doesn’t address async/performance issues
  2. Django REST Framework: Full-featured framework
    • Rejected: Too opinionated, heavy for our modular architecture
  3. Starlette: Direct use of ASGI framework
    • Rejected: Too low-level, would require building features FastAPI provides
  4. Tornado: Async web framework
    • Rejected: Less modern, smaller ecosystem
  5. Sanic: Fast async framework
    • Rejected: Less mature ecosystem, no automatic API documentation

Implementation Guidelines

Async by Default

class OrderController(BaseController):
    def __init__(self, order_service: OrderService, payment_service: PaymentService):
        super().__init__(prefix="/orders", tags=["orders"])
        self.order_service = order_service
        self.payment_service = payment_service
    
    @Get("/{order_id}")
    async def get_order(self, order_id: int) -> OrderResponse:
        # Async database call
        order = await self.order_service.get_order(order_id)
        
        # Async external service call
        payment = await self.payment_service.get_payment_status(order.payment_id)
        
        return OrderResponse(order=order, payment_status=payment.status)

class OrderFlavor(BaseFlavor):
    def __init__(self):
        super().__init__(
            controller=OrderController,
            services=[OrderService, PaymentService],
            repositories=[OrderRepository],
            config=OrderConfig
        )

Type Safety with Pydantic

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str
    age: int

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

class UserController(BaseController):
    def __init__(self, user_service: UserService):
        super().__init__(prefix="/users", tags=["users"])
        self.user_service = user_service
    
    @Post("/")
    async def create_user(self, user: UserCreate) -> UserResponse:
        # Automatic validation of user against UserCreate schema
        # Automatic serialization of response to UserResponse schema
        return await self.user_service.create_user(user)

Error Handling

from fastapi import HTTPException

class UserController(BaseController):
    def __init__(self, user_service: UserService):
        super().__init__(prefix="/users", tags=["users"])
        self.user_service = user_service
    
    @Get("/{user_id}")
    async def get_user(self, user_id: int) -> UserResponse:
        user = await self.user_service.get_user(user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user

References

I