ThaiWeather – Part 4: FastAPI Design - The Collector

from datetime import datetime, timezone
from typing import Any

import httpx
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

from app.models import City, WeatherObservation


class WeatherCollectionError(Exception):
    pass


class OpenMeteoCollector:
    SOURCE_NAME = "open-meteo"

    def __init__(
        self,
        api_url: str,
        timeout_seconds: float = 30,
    ) -> None:
        self.api_url = api_url
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(timeout_seconds),
            headers={
                "User-Agent": (
                    "ThaiWeatherCollector/1.0 "
                    "https://thaiweather.asia"
                )
            },
        )

    async def close(self) -> None:
        await self.client.aclose()

    @retry(
        retry=retry_if_exception_type(
            (
                httpx.TimeoutException,
                httpx.NetworkError,
                httpx.RemoteProtocolError,
            )
        ),
        stop=stop_after_attempt(3),
        wait=wait_exponential(
            multiplier=1,
            min=2,
            max=10,
        ),
        reraise=True,
    )
    async def collect(
        self,
        city: City,
    ) -> WeatherObservation:
        parameters = {
            "latitude": float(city.latitude),
            "longitude": float(city.longitude),
            "current": ",".join(
                [
                    "temperature_2m",
                    "relative_humidity_2m",
                    "surface_pressure",
                    "wind_speed_10m",
                    "wind_direction_10m",
                    "precipitation",
                    "cloud_cover",
                    "weather_code",
                ]
            ),
            "wind_speed_unit": "ms",
            "precipitation_unit": "mm",
            "timezone": "UTC",
        }

        response = await self.client.get(
            self.api_url,
            params=parameters,
        )

        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:
            raise WeatherCollectionError(
                f"Open Meteo returned HTTP "
                f"{response.status_code} for {city.name}"
            ) from exc

        try:
            payload = response.json()
        except ValueError as exc:
            raise WeatherCollectionError(
                f"Open Meteo returned invalid JSON for {city.name}"
            ) from exc

        if payload.get("error"):
            raise WeatherCollectionError(
                f"Open Meteo error for {city.name}: "
                f"{payload.get('reason', 'unknown error')}"
            )

        current = payload.get("current")

        if not isinstance(current, dict):
            raise WeatherCollectionError(
                f"No current weather returned for {city.name}"
            )

        measured_at = self._parse_datetime(
            current.get("time")
        )

        return WeatherObservation(
            city_id=city.id,
            temperature_c=self._to_float(
                current.get("temperature_2m")
            ),
            humidity=self._to_int(
                current.get("relative_humidity_2m")
            ),
            pressure_hpa=self._to_float(
                current.get("surface_pressure")
            ),
            wind_speed_ms=self._to_float(
                current.get("wind_speed_10m")
            ),
            wind_direction_deg=self._to_int(
                current.get("wind_direction_10m")
            ),
            rain_mm=self._to_float(
                current.get("precipitation")
            ),
            clouds_percent=self._to_int(
                current.get("cloud_cover")
            ),
            weather_code=self._to_string(
                current.get("weather_code")
            ),
            measured_at=measured_at,
            source=self.SOURCE_NAME,
        )

    @staticmethod
    def _parse_datetime(
        value: Any,
    ) -> datetime:
        if not isinstance(value, str):
            raise WeatherCollectionError(
                "Weather response contains no valid measurement time"
            )

        normalized_value = value.replace(
            "Z",
            "+00:00",
        )

        try:
            result = datetime.fromisoformat(
                normalized_value
            )
        except ValueError as exc:
            raise WeatherCollectionError(
                f"Invalid weather timestamp: {value}"
            ) from exc

        if result.tzinfo is None:
            result = result.replace(
                tzinfo=timezone.utc
            )

        return result.astimezone(timezone.utc)

    @staticmethod
    def _to_float(
        value: Any,
    ) -> float | None:
        if value is None:
            return None

        try:
            return float(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _to_int(
        value: Any,
    ) -> int | None:
        if value is None:
            return None

        try:
            return int(round(float(value)))
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _to_string(
        value: Any,
    ) -> str | None:
        if value is None:
            return None

        return str(value)

Flowchart

This class collects the current weather for one city from the Open Meteo API and converts the response into a WeatherObservation database model.

Imports

from datetime import datetime, timezone
from typing import Any

These are used to parse timestamps and accept values whose type is not known beforehand.

import httpx

httpx sends asynchronous HTTP requests. An asynchronous HTTP request lets your Python program do other work while it waits for a server to respond. With an asynchronous request, Python can temporarily handle other tasks during that waiting time:

from tenacity import retry, ...

Tenacity automatically retries the request when temporary network problems occur.

from app.models import City, WeatherObservation

These are your application models:

City contains coordinates and a city ID.
WeatherObservation represents the collected weather data.

Custom Exception

class WeatherCollectionError(Exception):
    pass

This defines an application specific exception.

Instead of exposing technical errors such as JSON parsing errors, the collector can raise meaningful errors such as:

Open Meteo returned invalid JSON for Bangkok

Other parts of your application can specifically catch:

except WeatherCollectionError:

The collector class

class OpenMeteoCollector:
    SOURCE_NAME = "open-meteo"

SOURCE_NAME is a class constant. Every observation created by this collector receives:

source="open-meteo"

That is useful if you later add another provider.

Initialising the HTTP client

def __init__(
    self,
    api_url: str,
    timeout_seconds: float = 30,
) -> None:

When you create the collector, you provide the API endpoint:

collector = OpenMeteoCollector(
    api_url="https://api.open-meteo.com/v1/forecast"
)

The constructor creates a reusable asynchronous HTTP client:

self.client = httpx.AsyncClient(...)

Reusing one client is more efficient than creating a new network connection for every city.

The timeout prevents the collector from waiting indefinitely:

timeout=httpx.Timeout(timeout_seconds)

The User-Agent tells Open Meteo which application is making the request:

"ThaiWeatherCollector/1.0 https://thaiweather.asia"

Closing the client

async def close(self) -> None:
    await self.client.aclose()

The client should be closed when collection finishes. This releases its network connections.

Example:

collector = OpenMeteoCollector(api_url)

try:
    observation = await collector.collect(city)
finally:
    await collector.close()

Automatic retry logic

The decorator applies retry behaviour to collect():

@retry(...)

It retries only these temporary connection problems:

httpx.TimeoutException
httpx.NetworkError
httpx.RemoteProtocolError

It makes at most three attempts:

stop=stop_after_attempt(3)

The waiting time increases exponentially:

wait_exponential(
    multiplier=1,
    min=2,
    max=10,
)

The delays will be approximately:

First failure: wait 2 seconds
Second failure: wait 2 seconds
Third failure: stop and raise the exception

reraise=True means the original network exception is raised after the final failed attempt.

Importantly, HTTP errors such as 404, 429, or 500 are not retried here. Only the specified network exceptions are retried.

Collecting weather for a city

async def collect(
    self,
    city: City,
) -> WeatherObservation:

The method receives a City object and returns a WeatherObservation.

Because it is asynchronous, it must be called using:

observation = await collector.collect(city)

Building the API parameters

parameters = {
    "latitude": float(city.latitude),
    "longitude": float(city.longitude),

Coordinates are converted to floats because database decimal fields may use another numeric type, such as Decimal.

The current parameter tells Open Meteo which fields you want:

"current": ",".join(
    [
        "temperature_2m",
        "relative_humidity_2m",
        ...
    ]
)

join() converts the list into this string:

temperature_2m,relative_humidity_2m,surface_pressure,...

The code explicitly request:

"wind_speed_unit": "ms",
"precipitation_unit": "mm",
"timezone": "UTC",

This is important because the database fields are called wind_speed_ms, rain_mm, and the timestamps are stored in UTC.

Sending the request

response = await self.client.get(
    self.api_url,
    params=parameters,
)

httpx adds the parameters to the URL as a query string.

Conceptually, it sends something like:

https://api.open-meteo.com/v1/forecast?latitude=13.75&longitude=100.50&timezone=UTC...

await allows Python to perform other asynchronous work while waiting for Open Meteo.

Checking the HTTP status

response.raise_for_status()

This raises httpx.HTTPStatusError when the server returns an unsuccessful HTTP status such as 400 or 500.

the code catches it and converts it into:

WeatherCollectionError(
    f"Open Meteo returned HTTP "
    f"{response.status_code} for {city.name}"
)

The syntax:

raise ... from exc

preserves the original exception as the cause. That helps with debugging and logging.

Parsing JSON

payload = response.json()

This converts the JSON response into Python dictionaries and values.

If Open Meteo returns malformed JSON, your code raises:

WeatherCollectionError(
    f"Open Meteo returned invalid JSON for {city.name}"
)

Checking for an API error

Some APIs return a valid JSON document containing an error:

{
  "error": true,
  "reason": "Invalid latitude"
}

This catches that situation:

if payload.get("error"):

If no reason is supplied, it uses:

unknown error

Extracting current weather

current = payload.get("current")

A successful response should contain something like:

{
  "current": {
    "time": "2026-08-18T10:00",
    "temperature_2m": 31.2,
    "relative_humidity_2m": 68
  }
}

This check confirms that current is a dictionary:

if not isinstance(current, dict):

Without it, later calls such as:

current.get("temperature_2m")

could crash.

Parsing the measurement time

measured_at = self._parse_datetime(
    current.get("time")
)

The helper method validates the timestamp and converts it to a timezone aware UTC datetime.

Creating the database model

return WeatherObservation(...)

This converts Open Meteo’s field names into our own domain model:

Open Meteo field	Your model field
`temperature_2m`	`temperature_c`
`relative_humidity_2m`	`humidity`
`surface_pressure`	`pressure_hpa`
`wind_speed_10m`	`wind_speed_ms`
`wind_direction_10m`	`wind_direction_deg`
`precipitation`	`rain_mm`
`cloud_cover`	`clouds_percent`
`weather_code`	`weather_code`
`time`	`measured_at`

The city is connected through:

city_id=city.id

This is the primary key to the cities table.

Notice that this code creates a model object, but it does not necessarily save it to PostgreSQL. The database session or repository still needs to add and commit it.

Timestamp conversion

normalized_value = value.replace(
    "Z",
    "+00:00",
)

ISO timestamps sometimes end with Z:

2026-08-18T10:00:00Z

Z means UTC. The code changes it to:

2026-08-18T10:00:00+00:00

It then parses the string:

result = datetime.fromisoformat(normalized_value)

If the API provides no timezone information:

if result.tzinfo is None:

the timestamp is assumed to be UTC:

result = result.replace(tzinfo=timezone.utc)

Finally, it guarantees that the result is expressed in UTC:

return result.astimezone(timezone.utc)

Safe value conversion

API responses are external input, so the values cannot be trusted completely.

return float(value)
return int(round(float(value)))
return str(value)

The values are converted to float, int or string and invalid or missing values become None.

ThaiWeather – Part 4: FastAPI Design – The Collector