Source code for action0.open_meteo.air_quality.operations

# generated by action0-client-openapi v0.1.2 from air-quality.yml — do not edit

from __future__ import annotations

import datetime
from typing import Any

from action0.client import JsonOperation
from action0.client import query
from action0.req import Method
from action0.req import Response

from .errors import BadRequestError
from .errors import decode_error
from .models import GetV1AirQualityCurrentItem
from .models import GetV1AirQualityDomains
from .models import GetV1AirQualityHourlyItem
from .models import GetV1AirQualityResponse
from .models import GetV1AirQualityTimeformat
from .models import get_v1_air_quality_error_from_json
from .models import get_v1_air_quality_response_from_json




class GetV1AirQuality(JsonOperation[GetV1AirQualityResponse]):
    """
    ``GET /v1/air-quality`` — Hourly air quality forecast

    Provides hourly air quality forecasts including PM2.5, PM10, ozone, nitrogen dioxide, pollen concentrations, and AQI indices from CAMS.
    """

    method = Method.GET
    path = "/v1/air-quality"

    latitude: str = query()
    longitude: str = query()
    #: A list of air quality variables which should be returned.
    hourly: list[GetV1AirQualityHourlyItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    #: A list of variables to return for the current conditions.
    current: list[GetV1AirQualityCurrentItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    timeformat: GetV1AirQualityTimeformat = query(default=GetV1AirQualityTimeformat.ISO8601)
    timezone: str | None = query(default=None)
    past_days: int = query(default=0)
    forecast_days: int = query(default=5)
    #: Start date in ISO 8601 format.
    start_date: datetime.date | None = query(default=None)
    #: End date in ISO 8601 format.
    end_date: datetime.date | None = query(default=None)
    #: CAMS domain: auto, cams_europe, or cams_global.
    domains: GetV1AirQualityDomains | None = query(default=None)
    #: Only required for commercial subscriptions.
    apikey: str | None = query(default=None)



    def check(self, response: Response) -> None:
        """
        :param response: the response the backend produced
        :raises BadRequestError: for the documented ``400`` answer
        :raises APIError: for any other non-2xx status
        """
        if response.status == 400:
            data = decode_error(response)
            if isinstance(data, dict):
                message = f"{type(self).__name__}: unexpected status {response.status}"
                raise BadRequestError(
                    f"{message} {response.phrase}".rstrip(),
                    response=response,
                    error=get_v1_air_quality_error_from_json(data),
                )
        super().check(response)




    def load_json(self, data: Any) -> GetV1AirQualityResponse:
        """
        :param data: the decoded JSON payload
        :return: the parsed result
        """
        return get_v1_air_quality_response_from_json(data)