FastAPI Tutorial

This document explains the FastAPI Weather API example provided in weather_api.py.

Overview

The example demonstrates how to:

1. Set up a FastAPI application
2. Define data models using Pydantic
3. Implement API endpoints for data ingestion and retrieval
4. Perform basic data processing
5. Use background tasks for asynchronous operations
6. Handle errors and implement logging

Code Explanation

Imports and Setup

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, Field
from typing import List
import logging
from datetime import datetime

app = FastAPI()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

weather_data = []
processed_data = []

This section imports necessary modules, sets up the FastAPI application, configures logging, and initializes in-memory storage for raw and processed data.

Data Models

class WeatherReading(BaseModel):
    station_id: str
    temperature: float = Field(..., ge=-50, le=60)
    humidity: float = Field(..., ge=0, le=100)
    wind_speed: float = Field(..., ge=0)
    timestamp: datetime

class ProcessedWeatherData(BaseModel):
    station_id: str
    avg_temperature: float
    avg_humidity: float
    max_wind_speed: float
    data_points: int

Here, we define Pydantic models for raw weather readings and processed weather data. The WeatherReading model includes data validation using Field.

Data Processing Function

def process_weather_data(station_id: str):
    station_data = [d for d in weather_data if d.station_id == station_id]
    if not station_data:
        return None

    avg_temp = sum(d.temperature for d in station_data) / len(station_data)
    avg_humidity = sum(d.humidity for d in station_data) / len(station_data)
    max_wind = max(d.wind_speed for d in station_data)

    processed = ProcessedWeatherData(
        station_id=station_id,
        avg_temperature=round(avg_temp, 2),
        avg_humidity=round(avg_humidity, 2),
        max_wind_speed=round(max_wind, 2),
        data_points=len(station_data)
    )
    processed_data.append(processed)
    logger.info(f"Processed data for station {station_id}")

This function processes raw weather data for a given station, calculating averages and maximum values.

API Endpoints

@app.post("/weather")
async def add_weather_data(reading: WeatherReading, background_tasks: BackgroundTasks):
    weather_data.append(reading)
    background_tasks.add_task(process_weather_data, reading.station_id)
    logger.info(f"Received data from station {reading.station_id}")
    return {"status": "Data received"}

@app.get("/weather/{station_id}", response_model=ProcessedWeatherData)
async def get_processed_data(station_id: str):
    data = next((d for d in processed_data if d.station_id == station_id), None)
    if not data:
        raise HTTPException(status_code=404, detail="Station data not found")
    return data

@app.get("/stations", response_model=List[str])
async def get_stations():
    return list(set(d.station_id for d in weather_data))

These endpoints handle:

1. Receiving weather data and triggering background processing
2. Retrieving processed data for a specific station
3. Listing all stations that have submitted data

Running the Application

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

This section allows the application to be run directly, starting the uvicorn server.

Running the Example

To run this example:

1. Ensure you have FastAPI and uvicorn installed:

   pip install fastapi uvicorn
   

2. Run the script:

   python fastapi_example.py
   

The API will be available at http://localhost:8001. You can interact with it using tools like cURL or access the automatic API documentation at http://localhost:8001/docs.

Examples

Here are some examples of how to use cURL to interact with each of the API endpoints in our FastAPI weather data example. These commands assume you're running the server locally on port 8001.

1. Adding weather data (POST /weather):

curl -X POST "http://localhost:8001/weather" \
     -H "Content-Type: application/json" \
     -d '{"station_id": "STAT001", "temperature": 25.5, "humidity": 60.0, "wind_speed": 10.2, "timestamp": "2023-09-22T14:30:00"}'

This command sends a POST request with a JSON payload containing weather data for station STAT001. You should receive a response like: {"status": "Data received"}

1. Getting processed data for a station (GET /weather/{station_id}):

curl "http://localhost:8001/weather/STAT001"

This retrieves the processed data for station STAT001. If data is available, you'll get a response like:

{
  "station_id": "STAT001",
  "avg_temperature": 25.5,
  "avg_humidity": 60.0,
  "max_wind_speed": 10.2,
  "data_points": 1
}

1. Getting a list of stations (GET /stations):

curl "http://localhost:8001/stations"

This will return a list of all station IDs that have submitted data: ["STAT001"]

Key Features Demonstrated

1. Data validation using Pydantic models
2. Asynchronous request handling
3. Background task processing
4. Error handling with HTTPException
5. Logging for monitoring and debugging
6. Automatic API documentation generation

key features and their significance in a data engineering context:

Data Models (WeatherReading and ProcessedWeatherData):

• Uses Pydantic for data validation and serialization.
• Ensures data integrity by defining field types and constraints.
• Significance: In data engineering, having well-defined data models is crucial for maintaining data quality and consistency throughout the pipeline.

Data Ingestion Endpoint (/weather POST):

• Receives weather data from multiple stations.
• Uses background tasks for asynchronous processing.
• Significance: Allows for efficient data ingestion, which is a common requirement in data engineering pipelines. The use of background tasks prevents blocking and improves API responsiveness.

Data Processing Function (process_weather_data):

• Aggregates and processes raw data.
• Calculates averages and maximum values.
• Significance: Demonstrates a simple data transformation process, which is a core aspect of data engineering. In real-world scenarios, this could involve more complex calculations or machine learning models.

Data Retrieval Endpoint (/weather/{station_id} GET):

• Provides access to processed data for each station.
• Implements error handling for missing data.
• Significance: Shows how processed data can be made available for consumption, which is often the end goal of data engineering pipelines.

Station List Endpoint (/stations GET):

• Provides a list of unique station IDs.
• Significance: Useful for data discovery and metadata management, which are important aspects of data engineering systems.

Logging:

• Implements basic logging for important events.
• Significance: Crucial for monitoring and debugging data pipelines in production environments.

Error Handling:

• Uses HTTPException for proper error responses.
• Significance: Ensures robust error handling, which is essential for maintaining the reliability of data pipelines.

Automatic API Documentation:

• FastAPI automatically generates OpenAPI (Swagger) documentation.
• Significance: Facilitates easy integration and usage of the API by other systems or teams, which is valuable in complex data engineering ecosystems.

Data Validation:

• Uses Pydantic fields with constraints (e.g., temperature range).
• Significance: Ensures data quality at the ingestion point, preventing invalid data from entering the pipeline.

This example showcases how FastAPI can be used to quickly create a robust API for data ingestion and processing, which is a common requirement in data engineering tasks.