Building a Weather Forecast API with FastAPI and PostgreSQL - Part 1 🌦️

When I set out to create a weather forecast API, I wanted something scalable, efficient, and built on modern technology. What started as a simple project to integrate weather data from the weather.gov API quickly grew into a complex, yet rewarding system. To keep things manageable, I decided to break this journey into multiple parts. In this post, I’ll cover the foundational steps to get started: setting up FastAPI, creating the database, and handling weather data with a robust schema.

The Problem ❓

I wanted to create a middle layer here for my weather data pipeline to handle a few things:

🌤️ Tracking historical forecasts.
📈 Storing multiple revisions of the same forecast.
🔍 Creating a self documenting data model with pydantic and redoc using the OpenAPI spec.

To handle these requirements, I needed a solution that could:

Fetch and store weather data efficiently.
Use a type-2 Slowly Changing Dimension (SCD) approach to track changes in forecasts over time.
Expose a well-documented, queryable REST API.

The Stack 🛠️

Here’s what I chose for this project:

FastAPI: A modern, high-performance web framework for Python. ⚡
PostgreSQL: A powerful relational database with excellent support for timestamps and schema design.
SQLAlchemy (Async): A robust ORM with support for asynchronous database queries.
Pydantic: For data validation and schema definition.
Docker: To containerize and simplify deployment.

Goals for Part 1 🎯

In this first post, we’ll:

Set up the FastAPI application structure.
Create a PostgreSQL database with a schema for storing weather forecasts.
Design a simple POST endpoint for adding forecast data.

Step 1: Setting Up the Project

Directory Structure

Here’s the modular structure I used for the FastAPI app:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
my_weather_api/
├── app/
│   ├── main.py                 # FastAPI entry point
│   ├── core/
│   │   ├── config.py           # Configuration settings (e.g., DB URL)
│   │   ├── database.py         # Database connection setup
│   ├── domains/
│   │   ├── weather/
│   │   │   ├── models.py       # SQLAlchemy models
│   │   │   ├── schemas.py      # Pydantic schemas
│   │   │   ├── repository.py   # Database logic (CRUD)
│   │   │   ├── router.py       # FastAPI endpoints
│   ├── __init__.py
├── .env                        # Environment variables (e.g., DB URL)
├── requirements.txt            # Python dependencies

Step 2: Database Schema 📋

Tracking Forecast Revisions

I wanted the ability to store multiple revisions of a forecast for the same time period. To do this, I used a composite primary key combining start_time and a unique id counter. This approach enables tracking of changes over time while preserving the original forecast data.

Here’s the SQLAlchemy model:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.dialects.postgresql import TIMESTAMP
from sqlalchemy.sql import func
from app.core.base import Base

class Forecast(Base):
    __tablename__ = "hourly_weather_forecast"

    start_time = Column(TIMESTAMP(timezone=True), primary_key=True, nullable=False, index=True)
    id = Column(Integer, primary_key=True, autoincrement=True)
    row_start_datetime = Column(TIMESTAMP(timezone=True), default=func.now(), nullable=False)
    row_end_datetime = Column(TIMESTAMP(timezone=True), nullable=True)
    is_current = Column(Boolean, default=True)
    end_time = Column(TIMESTAMP(timezone=True), nullable=False)
    temperature = Column(Integer)
    temperature_unit = Column(String)
    relative_humidity = Column(Integer)
    wind_speed = Column(String)
    wind_direction = Column(String)
    short_forecast = Column(String)
    icon = Column(String)

Step 3: Adding Data to the API

Pydantic Schema

To validate incoming data, I used a Pydantic schema that matches the database structure:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from pydantic import BaseModel
from datetime import datetime

class ForecastCreate(BaseModel):
    start_time: datetime
    end_time: datetime
    temperature: int
    temperature_unit: str
    relative_humidity: int
    wind_speed: str
    wind_direction: str
    short_forecast: str
    icon: str

FastAPI Endpoint

The /forecast endpoint allows you to add new forecast data:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.domains.weather.repository import upsert_forecast_data
from app.domains.weather.schemas import ForecastCreate

router = APIRouter()

@router.post("/", response_model=ForecastCreate)
async def create_forecast(
    forecast_data: ForecastCreate, db: AsyncSession = Depends(get_db)
):
    try:
        return await upsert_forecast_data(db, forecast_data)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error creating forecast: {e}")

Step 4: Testing the API

Inserting Data

Using curl:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
curl -X POST "http://127.0.0.1:8000/forecast/" -H "Content-Type: application/json" -d '{
  "start_time": "2024-11-26T15:00:00Z",
  "end_time": "2024-11-26T18:00:00Z",
  "temperature": 72,
  "temperature_unit": "Fahrenheit",
  "relative_humidity": 45,
  "wind_speed": "15 mph",
  "wind_direction": "NE",
  "short_forecast": "Partly Cloudy",
  "icon": "https://example.com/icons/partly-cloudy.png"
}'

Expected Response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
{
  "start_time": "2024-11-26T15:00:00Z",
  "end_time": "2024-11-26T18:00:00Z",
  "temperature": 72,
  "temperature_unit": "Fahrenheit",
  "relative_humidity": 45,
  "wind_speed": "15 mph",
  "wind_direction": "NE",
  "short_forecast": "Partly Cloudy",
  "icon": "https://example.com/icons/partly-cloudy.png"
}

What’s Next?

In Part 2, we’ll:

Build query endpoints to fetch current forecasts or historical revisions.
Discuss how to efficiently implement a type-2 SCD for this API.
Handle advanced use cases like filtering and pagination.

This sets a solid foundation for a scalable weather API. Stay tuned for the next part! 🌟

The Problem ❓#

The Stack 🛠️#

Goals for Part 1 🎯#

Step 1: Setting Up the Project#

Directory Structure#

Step 2: Database Schema 📋#

Tracking Forecast Revisions#

Step 3: Adding Data to the API#

Pydantic Schema#

FastAPI Endpoint#

Step 4: Testing the API#

Inserting Data#

Expected Response#

What’s Next?#