xbe.at

Web and APIs

FastAPI with Python

6 min read, 15 sections

Contents
  1. Introduction to FastAPI
  2. Key Features of FastAPI
  3. Path Parameters
  4. Query Parameters
  5. Request Body
  6. Response Models
  7. Dependency Injection
  8. Background Tasks
  9. WebSockets
  10. Error Handling
  11. Middleware
  12. Testing FastAPI Applications
  13. Real-Life Example: Task Management API
  14. Real-Life Example: Weather Information API
  15. Additional Resources

Introduction to FastAPI

FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. It's designed to be easy to use, fast to code, ready for production, and capable of handling high loads.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

# To run: uvicorn main:app --reload

Key Features of FastAPI

FastAPI offers automatic API documentation, data validation, serialization, and more. It's built on top of Starlette for the web parts and Pydantic for the data parts, combining speed and simplicity.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}

Path Parameters

Path parameters allow you to capture values from the URL. FastAPI automatically validates and converts the parameters to the specified type.

from fastapi import FastAPI

app = FastAPI()

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

# Example URL: http://localhost:8000/items/42
# Result: {"item_id": 42}

Query Parameters

Query parameters are key-value pairs in the URL after the ? symbol. FastAPI automatically parses and validates these parameters.

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# Example URL: http://localhost:8000/items/?skip=20&limit=50
# Result: {"skip": 20, "limit": 50}

Request Body

FastAPI uses Pydantic models to define the structure of request bodies, providing automatic validation and serialization.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str
    full_name: str = None

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

# Example request body:
# {
#     "username": "johndoe",
#     "email": "[email protected]",
#     "full_name": "John Doe"
# }

Response Models

Response models define the structure of the API responses, ensuring type safety and automatic documentation.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

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

# This ensures the response matches the Item model

Dependency Injection

FastAPI's dependency injection system allows you to declare shared logic or data processing as dependencies, promoting code reuse and separation of concerns.

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")
    return user

@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

# This example demonstrates a simple authentication dependency

Background Tasks

FastAPI allows you to define background tasks that run after returning a response, ideal for operations that don't need to block the response.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

WebSockets

FastAPI supports WebSockets, allowing real-time bidirectional communication between the client and the server.

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

# This creates a WebSocket endpoint that echoes received messages

Error Handling

FastAPI provides a straightforward way to handle errors and exceptions, allowing you to return custom error responses.

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

# This returns a 404 error if the item is not found

Middleware

Middleware allows you to add custom functionality to the request/response cycle, such as CORS, authentication, or logging.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

# This adds CORS middleware to allow all origins

Testing FastAPI Applications

FastAPI is built on top of Starlette, which provides a TestClient for easy testing of your API endpoints.

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

# This sets up a simple test for the root endpoint

Real-Life Example: Task Management API

This example demonstrates a simple task management API using FastAPI, showcasing CRUD operations.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Task(BaseModel):
    id: Optional[int] = None
    title: str
    description: Optional[str] = None
    completed: bool = False

tasks = []

@app.post("/tasks/", response_model=Task)
async def create_task(task: Task):
    task.id = len(tasks) + 1
    tasks.append(task)
    return task

@app.get("/tasks/", response_model=List[Task])
async def read_tasks():
    return tasks

@app.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: int):
    task = next((task for task in tasks if task.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, updated_task: Task):
    task = next((task for task in tasks if task.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    task.title = updated_task.title
    task.description = updated_task.description
    task.completed = updated_task.completed
    return task

@app.delete("/tasks/{task_id}", response_model=Task)
async def delete_task(task_id: int):
    task = next((task for task in tasks if task.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    tasks.remove(task)
    return task

# This API allows creating, reading, updating, and deleting tasks

Real-Life Example: Weather Information API

This example showcases a simple weather information API using FastAPI, demonstrating how to work with external data sources and caching.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from cachetools import TTLCache

app = FastAPI()

# Simple in-memory cache with a 30-minute TTL
cache = TTLCache(maxsize=100, ttl=1800)

class WeatherInfo(BaseModel):
    city: str
    temperature: float
    description: str

@app.get("/weather/{city}", response_model=WeatherInfo)
async def get_weather(city: str):
    if city in cache:
        return cache[city]

    # Simulating an external API call
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/weather/{city}")
        if response.status_code != 200:
            raise HTTPException(status_code=404, detail="City not found")
        data = response.json()

    weather_info = WeatherInfo(
        city=city,
        temperature=data["temperature"],
        description=data["description"]
    )

    # Cache the result
    cache[city] = weather_info
    return weather_info

# This API fetches weather information for a given city and caches the results

Additional Resources

For more information on FastAPI and related topics, consider exploring these resources:

  1. FastAPI Official Documentation: https://fastapi.tiangolo.com/
  2. Starlette Documentation: https://www.starlette.io/
  3. Pydantic Documentation: https://pydantic-docs.helpmanual.io/
  4. "Asynchronous Web APIs with FastAPI" on ArXiv: https://arxiv.org/abs/2108.03261
  5. "A Comparative Analysis of FastAPI and Flask for Building RESTful APIs" on ArXiv: https://arxiv.org/abs/2204.10584

These resources provide in-depth information on FastAPI, its underlying technologies, and comparisons with other frameworks.

More in Web and APIs

  1. Introduction to Flask-SQLAlchemy3 min

    Mastering Flask-SQLAlchemy for Efficient Database Management

  2. Django Two-Factor Authentication for Secure User Login7 min

    Django two-factor authentication enhances security by requiring users to provide additional verification beyond passwords. The django-two-factor-auth package...

  3. Lasso L1 Penalty in Machine Learning with Python11 min

    Lasso (Least Absolute Shrinkage and Selection Operator) is a regularization technique in machine learning that uses L1 penalty. It's particularly useful for...

  4. Django REST Framework Serialization Fundamentals5 min

    The Django Rest Framework serializer acts as a converter between Python objects and JSON/XML formats. It provides validation, data formatting, and model...

  5. Lasso Regression in scikit-learn6 min

    Linear regression with L1 regularization, also known as Lasso (Least Absolute Shrinkage and Selection Operator), adds a penalty term to the loss function...

  6. Django REST Framework Filtering Techniques4 min

    Django Rest Framework filtering requires initial configuration and package installation. The django-filter package extends DRF's filtering capabilities by...