Skip to content

Base Weather

SparkWeatherCompanyBaseWeatherSource

Bases: BaseISOSource

Base class for all the Weather related sources. Provides common functionality.

Parameters:

Name Type Description Default
spark SparkSession

Spark Session instance

 required
options dict

A dictionary of Weather Source specific configurations.

 required
Source code in src/sdk/python/rtdip_sdk/pipelines/sources/spark/the_weather_company/base_weather.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class SparkWeatherCompanyBaseWeatherSource(BaseISOSource):
    """
    Base class for all the Weather related sources. Provides common functionality.

    Parameters:
        spark (SparkSession): Spark Session instance
        options (dict): A dictionary of Weather Source specific configurations.

    """

    spark: SparkSession
    options: dict
    weather_url: str = "https://"
    api_params: dict = {}

    def __init__(self, spark: SparkSession, options: dict) -> None:
        super(SparkWeatherCompanyBaseWeatherSource, self).__init__(spark, options)
        self.spark = spark
        self.options = options

    def _get_api_params(self) -> dict:
        return self.api_params

    def _fetch_weather_from_url(self, url_suffix: str, params: dict) -> bytes:
        """
        Gets data from external Weather Forecast API.

        Args:
            url_suffix: String to be used as suffix to weather url.

        Returns:
            Raw content of the data received.

        """
        url = f"{self.weather_url}{url_suffix}"
        logging.info(f"Requesting URL - `{url}` with params - {params}")

        response = requests.get(url, params)
        code = response.status_code

        if code != 200:
            raise HTTPError(
                f"Unable to access URL `{url}`."
                f" Received status code {code} with message {response.content}"
            )
        return response.content

    def _fetch_from_url(self, url_suffix: str) -> bytes:
        return self._fetch_weather_from_url(url_suffix, self._get_api_params())