Building the weather cache

Time to combine everything from the chapter into the artefact the overview promised. weather_cache.py wraps the schema, the freshness check, the storage, and the analytics queries inside a single WeatherCache class.

Application code calls cache.get_weather(location). The class either returns a recent row from SQLite or fetches fresh data from the API and stores it. The caller uses the same method either way. That is the point of the cache: persistence becomes an implementation detail behind a simple Python interface.

The cache flow

Three behaviours separate a cache from a plain "save the response" wrapper:

  • Read before fetching. Every get_weather(location) call first checks the database for a row newer than the freshness window. A hit returns immediately, no network.
  • Store after fetching. A miss calls the API, then writes the result back so the next call within the window is a hit.
  • Answer historical questions. API misses are stored as new rows, so readings accumulate over time. That lets the cache surface daily averages, recent trends, and "last 7 days" reports without making extra API calls for old data.

That's the whole feature set. Before you read the full class, here is the flow:

YES NO CALL get_weather(location) CHECK _fetch_cached() asks SQLite for a fresh row Fresh row in cache? CACHE HIT Return the stored row. No network call. source = "cache" CACHE MISS Call the API, store the new reading, return it. source = "api"
served from database network fetch

The important detail is that both branches return the same public shape: a weather-reading dict with database fields such as id and recorded_at, plus a source value telling you whether the row came from the cache or the network.

Before you run this

This page calls the OpenWeatherMap API. It assumes requests and python-dotenv are installed, and that OPENWEATHER_API_KEY is saved in your local .env file. You do not need to create weather.db first: this class creates the database, table, and index when WeatherCache starts.

The implementation below wraps the SQL idioms from the previous pages. Read it in four passes:

  1. _init_schema() creates the table and index.
  2. _fetch_cached() checks for a fresh row.
  3. _fetch_from_api() and _store() handle cache misses.
  4. get_weather() ties the decision together.

Save this as weather_cache.py at the project root:

weather_cache.py 
"""
WeatherCache: a SQLite-backed cache for OpenWeatherMap responses.
"""
import os
import sqlite3
from contextlib import closing

import requests
from dotenv import load_dotenv

API_URL = "https://api.openweathermap.org/data/2.5/weather"


load_dotenv()


class WeatherCache:
    def __init__(self, db_path="weather.db", api_key=None, freshness_minutes=60):
        self.db_path = db_path
        self.api_key = api_key or os.environ.get("OPENWEATHER_API_KEY")
        self.freshness_minutes = freshness_minutes
        if not self.api_key:
            raise RuntimeError(
                "OPENWEATHER_API_KEY missing. Set it in .env "
                "or pass api_key= to WeatherCache()."
            )
        self._init_schema()

    def _connect(self):
        """Open a connection that closes when its `with` block exits."""
        return closing(sqlite3.connect(self.db_path))

    @staticmethod
    def _row_to_dict(row):
        return dict(row) if row is not None else None

    def _init_schema(self):
        with self._connect() as conn:
            with conn:
                conn.execute(
                    """
                    CREATE TABLE IF NOT EXISTS weather_readings (
                        id           INTEGER PRIMARY KEY,
                        location     TEXT NOT NULL,
                        temperature  REAL,
                        conditions   TEXT,
                        humidity     INTEGER,
                        wind_speed   REAL,
                        recorded_at  DATETIME DEFAULT CURRENT_TIMESTAMP
                    )
                    """
                )
                conn.execute(
                    """
                    CREATE INDEX IF NOT EXISTS idx_weather_location_recorded_at
                    ON weather_readings(location, recorded_at DESC)
                    """
                )

    def _fetch_cached(self, location):
        """Return the most recent fresh reading for `location`, or None."""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(
                """
                SELECT *
                FROM weather_readings
                WHERE location = ?
                  AND recorded_at > datetime('now', ?)
                ORDER BY recorded_at DESC
                LIMIT 1
                """,
                (location, f"-{self.freshness_minutes} minutes"),
            )
            return self._row_to_dict(cursor.fetchone())

    def _fetch_from_api(self, location):
        """Hit OpenWeatherMap and return a normalised reading dict."""
        response = requests.get(
            API_URL,
            params={"q": location, "appid": self.api_key, "units": "metric"},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()
        return {
            "location": location,
            "temperature": data["main"]["temp"],
            "conditions": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"],
            "wind_speed": data["wind"]["speed"],
        }

    def _store(self, reading):
        # recorded_at uses SQLite's default timestamp format.
        # That keeps the freshness check compatible with datetime('now', ...).
        with self._connect() as conn:
            with conn:
                cursor = conn.execute(
                    """
                    INSERT INTO weather_readings
                        (location, temperature, conditions, humidity, wind_speed)
                    VALUES (?, ?, ?, ?, ?)
                    """,
                    (
                        reading["location"],
                        reading["temperature"],
                        reading["conditions"],
                        reading["humidity"],
                        reading["wind_speed"],
                    ),
                )
                return cursor.lastrowid

    def _fetch_by_id(self, reading_id):
        """Return a stored reading by primary key."""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(
                """
                SELECT *
                FROM weather_readings
                WHERE id = ?
                """,
                (reading_id,),
            )
            return self._row_to_dict(cursor.fetchone())

    def get_weather(self, location):
        """Return weather for `location`, from cache if fresh, else from API."""
        cached = self._fetch_cached(location)
        if cached is not None:
            cached["source"] = "cache"
            return cached

        reading = self._fetch_from_api(location)
        reading_id = self._store(reading)
        stored = self._fetch_by_id(reading_id)
        if stored is None:
            raise RuntimeError("Stored weather reading could not be found.")
        stored["source"] = "api"
        return stored

    def daily_averages(self, location, days=7):
        """Average temperature per day for the last `days` days."""
        with self._connect() as conn:
            cursor = conn.execute(
                """
                SELECT date(recorded_at) AS day,
                       AVG(temperature)  AS avg_temp,
                       COUNT(*)          AS readings
                FROM weather_readings
                WHERE location = ?
                  AND recorded_at >= date('now', ?)
                GROUP BY date(recorded_at)
                ORDER BY day
                """,
                (location, f"-{days} days"),
            )
            return [
                {"day": day, "avg_temp": avg_temp, "readings": readings}
                for day, avg_temp, readings in cursor
            ]

How the class fits together

The setup path

The class has three small infrastructure helpers. _connect() opens a SQLite connection wrapped in contextlib.closing(), so each with self._connect() as conn: block closes the connection when it exits.

The nested with conn: blocks inside write methods handle commit and rollback. _row_to_dict() converts sqlite3.Row objects into ordinary dictionaries before returning them to the rest of the class.

_init_schema() is deliberately inside the class, even though earlier pages used weather_schema.sql. A schema file is a good teaching and setup tool; application code often also needs an initialisation path so a new environment can prepare its own database automatically. Here, constructing WeatherCache is enough to make sure the table and freshness-check index exist.

The cache-hit path

The cache path is deliberately small. _fetch_cached() uses the same SELECT, WHERE, ORDER BY, and LIMIT pattern from the querying page, returning the latest reading inside the freshness window or None. get_weather() calls it first and marks hits with source = "cache".

The cache-miss path

The miss path is the only place the network is touched. _fetch_from_api() calls OpenWeatherMap and normalises the JSON into the fields this app stores. _store() writes the reading and returns the new row's id. _fetch_by_id() then reads that stored row back so API misses return the same shape as cache hits, including id and recorded_at.

The public interface

The two public entry points are get_weather() (the cache itself) and daily_averages() (the analytics query that justifies storing every reading). Both hide the SQL behind a Python interface; calling code never has to write a query.

Run the cache

The first run creates weather.db, creates the weather_readings table, and adds the freshness-check index. Save this as use_cache.py alongside weather_cache.py:

use_cache.py 
from weather_cache import WeatherCache

cache = WeatherCache(freshness_minutes=60)

# First call: cache miss, hits the API, stores the row.
reading = cache.get_weather("Dublin")
print(f"[{reading['source']}] Dublin: {reading['temperature']}°C, {reading['conditions']}")

# Second call within the hour: cache hit, no network.
reading = cache.get_weather("Dublin")
print(f"[{reading['source']}] Dublin: {reading['temperature']}°C, {reading['conditions']}")

# Historical view across whatever's been collected.
print("\nDaily averages (last 7 days):")
for row in cache.daily_averages("Dublin", days=7):
    label = "reading" if row["readings"] == 1 else "readings"
    print(f"  {row['day']}: {row['avg_temp']:.1f}°C ({row['readings']} {label})")

Add your API key to .env:

.env 
OPENWEATHER_API_KEY=your_key_here

Then run the script:

Terminal 
python use_cache.py

The output should look like this:

Output 
[api] Dublin: 12.5°C, cloudy
[cache] Dublin: 12.5°C, cloudy

Daily averages (last 7 days):
  2026-05-06: 12.5°C (1 reading)

The exact temperature and conditions will vary because OpenWeatherMap returns live data. The important part is the source field: the first get_weather("Dublin") call should return "api", and the second call should return "cache". Run the script several times across an hour and the call counts shift over to mostly cache hits, with a single API call per location per freshness window.

Where the cache decisions live

A few places in the code carry decisions worth naming, because they're the levers you'd tweak when adapting this for a different API:

  • The freshness window. freshness_minutes=60 is the "how recent counts as fresh" knob. For weather, an hour is reasonable. For stock prices it might be 60 seconds; for repository metadata it might be 24 hours. The freshness check is just WHERE recorded_at > datetime('now', '-N minutes'); the constant is the only thing that changes.
  • Append vs upsert. The schema here appends a new row every time the API is called, so historical analytics are possible. If you only care about "the latest value per location" and do not need history, use the weather_cache shape from the inserting rows page, with location as the primary key and an ON CONFLICT(location) DO UPDATE upsert. Pick based on whether you want history.
  • Storing the raw API response. This implementation extracts a few fields from the JSON. For projects where the API's response shape might change, store the entire JSON as a TEXT column too (raw_response), so old data is recoverable even if your extraction logic gets out of step with the API.
  • Index choice. The composite index on (location, recorded_at DESC) is the one the freshness check uses. If you start querying by other patterns (say, "all readings between two timestamps regardless of location"), you'll want a different index.

What this gives you

Three things you couldn't do with a fetch-only script that you can do now:

  • Run while offline if cached data is fresh enough.
  • Survive restarts; readings persist across runs because they're on disk, not in memory.
  • Inspect the raw data with the SQLite CLI.
Terminal 
sqlite3 weather.db "SELECT * FROM weather_readings ORDER BY recorded_at DESC LIMIT 10"

The next two pages step back from the worked example and examine the design choices behind it: schema design, safe schema changes, index selection, query plans, and the signals that tell you when SQLite is no longer the right storage layer.