FastAPI vs. AIOHTTP: Comparison for Python Developers

FastAPI and AIOHTTP are two popular Python frameworks for building asynchronous web applications.

FastAPI is known for its ease of use, and automatic API documentation, while AIOHTTP offers more flexibility and lower-level control.

This tutorial will compare these frameworks across various aspects to help you choose the right tool for your project.

 

 

Benchmark Comparisons

To compare the performance of FastAPI and AIOHTTP, you can use a simple benchmark test that measures the time taken to handle a set number of requests.

I created 2 simple servers using FastAPI and AIOHTTP to perform the benchmark, the FastAPI server is on port 8000 and the AIOHTTP server is on port 8089.

import time
import requests
def benchmark(url, num_requests=1000):
  start_time = time.time()
  for _ in range(num_requests):
      requests.get(url)
  end_time = time.time()
  return end_time - start_time
fastapi_time = benchmark('http://localhost:8000')
aiohttp_time = benchmark('http://localhost:8090')
print(f"FastAPI: {fastapi_time} seconds")
print(f"AIOHTTP: {aiohttp_time} seconds")

Output:

FastAPI: 20.461241960525513 seconds
AIOHTTP: 20.342910051345825 seconds

They show same performance for a simple GET request.

 

Type Checking and Validation

FastAPI Integration with Pydantic

FastAPI uses Pydantic for data validation and type checking to define and validate request bodies.

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
  name: str
  price: float

@app.post("/items/")
async def create_item(item: Item):
  return item

Output:

{"name": "Widget", "price": 19.99}

FastAPI automatically validates the request body against the Pydantic model.

AIOHTTP Requires Validation Libraries

AIOHTTP requires additional libraries like pydantic or marshmallow for data validation.

from aiohttp import web
from pydantic import BaseModel, ValidationError
class Item(BaseModel):
  name: str
  price: float
async def handle(request):
  try:
      data = await request.json()
      item = Item(**data)
      return web.json_response(item.dict())
  except ValidationError as e:
      return web.json_response({"error": str(e)}, status=400)
app = web.Application()
app.router.add_post('/items/', handle)
if __name__ == '__main__':
  web.run_app(app)

Output:

{"name": "Widget", "price": 19.99}

AIOHTTP requires manual validation using Pydantic which makes it complex compared to FastAPI built-in support.

 

Request Body Parsing

FastAPI Automatic Request Body Parsing

FastAPI automatically parses request bodies based on the defined Pydantic models.

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class User(BaseModel):
  username: str
  email: str

@app.post("/users/")
async def create_user(user: User):
  return user

Output:

{"username": "Adam", "email": "adam@example.com"}

FastAPI simplifies request body parsing by automatically converting JSON payloads into Pydantic model instances.

AIOHTTP Manual Body Parsing

AIOHTTP requires manual parsing of request bodies, typically using json() method.

from aiohttp import web
async def handle(request):
  data = await request.json()
  return web.json_response(data)
app = web.Application()
app.router.add_post('/users/', handle)
if __name__ == '__main__':
  web.run_app(app)

Output:

{"username": "Adam", "email": "adam@example.com"}

AIOHTTP requires explicit parsing of request bodies, which can be more verbose than automatic handling of FastAPI.

 

Response Formatting

FastAPI Automatic Response Serialization

FastAPI automatically serializes responses to JSON, simplifying the response creation process.

from fastapi import FastAPI
app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
  return {"item_id": item_id, "name": "Widget"}

Output:

{"item_id": 1, "name": "Widget"}

AIOHTTP Manual Response Formatting

AIOHTTP requires manual response formatting using json_response.

from aiohttp import web
async def handle(request):
  item_id = request.match_info.get('item_id', "Anonymous")
  return web.json_response({"item_id": item_id, "name": "Widget"})
app = web.Application()
app.router.add_get('/items/{item_id}', handle)
if __name__ == '__main__':
  web.run_app(app)

Output:

{"item_id": "1", "name": "Widget"}

AIOHTTP requires explicit response formatting, which can be more cumbersome compared to FastAPI automatic serialization.

 

WebSocket Support

FastAPI: Native WebSocket Support

FastAPI offers native support for WebSockets to implement real-time communication.

from fastapi import FastAPI, WebSocket
app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
  await websocket.accept()
  await websocket.send_text("Hello WebSocket")
  await websocket.close()

Output:

WebSocket message: Hello WebSocket

AIOHTTP Requires More Configuration

AIOHTTP requires additional configuration to support WebSockets.

from aiohttp import web
async def websocket_handler(request):
  ws = web.WebSocketResponse()
  await ws.prepare(request)
  await ws.send_str("Hello WebSocket")
  await ws.close()
  return ws
app = web.Application()
app.router.add_get('/ws', websocket_handler)
if __name__ == '__main__':
  web.run_app(app)

Output:

WebSocket message: Hello WebSocket

AIOHTTP’s WebSocket support requires more manual setup, which can be more complex compared to FastAPI straightforward approach.

 

Authentication and Authorization

FastAPI: Provides Built-in Support

FastAPI offers built-in support for authentication and authorization using OAuth2 and JWT.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
  if token != "fake-token":
      raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
  return {"username": "Adam"}

Output:

{"username": "Adam"}

AIOHTTP Requires Custom Implementations

AIOHTTP requires custom implementations or third-party libraries for authentication.

from aiohttp import web
async def handle(request):
  token = request.headers.get('Authorization')
  if token != "Bearer fake-token":
      return web.json_response({"error": "Invalid token"}, status=401)
  return web.json_response({"username": "Adam"})
app = web.Application()
app.router.add_get('/users/me', handle)

Output:

{"username": "Adam"}

AIOHTTP requires more manual setup for authentication, which can be more complex compared to FastAPI built-in features.

 

Low Level Access (AIOHTTP Advantage)

In web frameworks, low-level access refers to the ability to interact directly with the HTTP protocol and manage the details of request and response handling.

This includes manipulating headers, managing connections, handling raw data streams, and implementing custom protocols.

Direct Manipulation of HTTP Requests and Responses

AIOHTTP allows you to directly access and modify HTTP requests and responses, providing more control over the data flow.

from aiohttp import web
async def handle(request):
  # Accessing request headers
  user_agent = request.headers.get('User-Agent', 'Unknown')
  
  # Create custom response
  response = web.Response(text=f"Hello, your User-Agent is {user_agent}")
  
  # Modify response headers
  response.headers['X-Custom-Header'] = 'CustomValue'  
  return response
app = web.Application()
app.router.add_get('/', handle)
web.run_app(app)

Output:

Hello, your User-Agent is <your-user-agent>

This code shows how you can directly access request headers and modify response headers.

This level of customization is more cumbersome in FastAPI.

Custom Handling of Data Streams

AIOHTTP supports streaming data, which is useful for handling large files or real-time data.

from aiohttp import web
async def stream_response(request):
  response = web.StreamResponse()
  response.headers['Content-Type'] = 'text/plain'
  await response.prepare(request)
  for i in range(10):
      await response.write(f"Chunk {i}\n".encode('utf-8'))
      await asyncio.sleep(1)
  return response
app = web.Application()
app.router.add_get('/stream', stream_response)
web.run_app(app)

Output:

Chunk 0
Chunk 1
Chunk 2
Chunk 3
Chunk 4
Chunk 5
Chunk 6
Chunk 7
Chunk 8
Chunk 9

This example demonstrates AIOHTTP’s ability to handle streaming responses, which is more complex to implement in FastAPI due to its higher-level abstractions.

Implement Custom Protocols

AIOHTTP allows you to implement custom protocols by providing low-level access to the connection lifecycle.

from aiohttp import web
async def custom_protocol_handler(request):
  # Custom protocol logic
  data = await request.read()
  return web.Response(text=f"Received data: {data.decode('utf-8')}")
app = web.Application()
app.router.add_post('/custom', custom_protocol_handler)
web.run_app(app)

Output:

Received data: <your-data>