ims_envista.ims_envista.IMSEnvista.__init__() - Code Metrics - Inspection of "Reduce Scurtinizer warnings (#7)" - GuyKh/py-ims-envista - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 5bd916...7cb3ea )

by Guy

created 2023-03-01 09:57 UTC

ims_envista.ims_envista.IMSEnvista.init() A

↳ Parent: ims_envista.ims_envista

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	2
dl	0
loc	6
rs	10
c	0
b	0
f	0

"""Module IMSEnvista getting IMS meteorological readings."""

from __future__ import annotations
import json
from typing import Optional, List

from datetime import date
import requests
from requests.adapters import HTTPAdapter
from requests.exceptions import HTTPError
from loguru import logger
from urllib3 import Retry

from .const import (
    GET_LATEST_STATION_DATA_URL,
    GET_EARLIEST_STATION_DATA_URL,
    GET_STATION_DATA_BY_DATE_URL,
    GET_SPECIFIC_STATION_DATA_URL,
    GET_ALL_STATIONS_DATA_URL,
    GET_ALL_REGIONS_DATA_URL,
    API_REGION_ID,
    API_NAME,
    API_STATIONS,
    GET_SPECIFIC_REGION_DATA_URL,
    GET_DAILY_STATION_DATA_URL,
    GET_MONTHLY_STATION_DATA_URL,
    GET_MONTHLY_STATION_DATA_BY_MONTH_URL,
    GET_STATION_DATA_BY_RANGE_URL,
    VARIABLES
)
from .ims_variable import IMSVariable
from .meteo_data import (
    station_meteo_data_from_json,
    StationMeteorologicalReadings,
)
from .station_data import StationInfo, station_from_json, region_from_json, RegionInfo

# ims.gov.il does not support ipv6 yet, `requests` use ipv6 by default
# and wait for timeout before trying ipv4, so we have to disable ipv6
requests.packages.urllib3.util.connection.HAS_IPV6 = False

def create_session():

    session = requests.Session()
    retry = Retry(connect=3, backoff_factor=0.5)
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session


class IMSEnvista:
    """API Wrapper to IMS Envista"""

    def __init__(self, token: str):
        if not token:
            raise ValueError

        self.token = token
        self.session = create_session()

    @staticmethod
    def _get_channel_id_url_part(channel_id: int) -> str:
        """get specific Channel Id url param"""
        if channel_id:
            return "/" + str(channel_id)
        return ""

    def _get_ims_url(self, url: str) -> Optional[dict]:
        """Fetches data from IMS url

        Args:
            url (str): IMS Station ID

        Returns:
            data: Current station meteorological data
        """
        logger.debug(f"Fetching data from: {url}")
        try:
            response = self.session.get(
                url,
                headers={
                    "Accept": "application/vnd.github.v3.text-match+json",
                    "Authorization": f"ApiToken {self.token}",
                },
                timeout=10,
            )

            # If the response was successful, no Exception will be raised
            response.raise_for_status()
            return json.loads(response.text)
        except HTTPError as http_err:
            logger.error(f"HTTP error occurred: {http_err}")  # Python 3.6
            return None
        except Exception as err:

            logger.error(f"Other error occurred: {err}")  # Python 3.6
            return None

    def close(self):
        """ Close Requests Session """
        self.session.close()

    def get_latest_station_data(
            self, station_id: int, channel_id: int = None
        ) -> StationMeteorologicalReadings:
        """Fetches the latest station data from IMS Envista API

        Args:
            station_id (int): IMS Station Id
            channel_id (int): [Optional] Specific Channel Id

        Returns:
            data: Current station meteorological data
        """
        get_url = GET_LATEST_STATION_DATA_URL.format(
            str(station_id), self._get_channel_id_url_part(channel_id)
        )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_earliest_station_data(
            self, station_id: int, channel_id: int = None
        ) -> StationMeteorologicalReadings:
        """Fetches the earliest station data from IMS Envista API

        Args:
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel ID

        Returns:
            data: Current station meteorological data
        """
        get_url = GET_EARLIEST_STATION_DATA_URL.format(
            str(station_id), self._get_channel_id_url_part(channel_id)
        )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_station_data_from_date(
            self, station_id: int, date_to_query: date, channel_id: int = None
        ) -> StationMeteorologicalReadings:
        """Fetches latest station data from IMS Envista API by date

        Args:
            station_id (int): IMS Station ID
            date_to_query (date): Selected date to query
            channel_id (int): [Optional] Specific Channel Id

        Returns:
            data: Current station meteorological data
        """
        get_url = GET_STATION_DATA_BY_DATE_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
            str(date_to_query.year),
            str(date_to_query.month),
            str(date_to_query.day),
        )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_station_data_by_date_range(
            self,
            station_id: int,
            from_date: date,
            to_date: date,
            channel_id: int = None,
        ) -> StationMeteorologicalReadings:
        """Fetches latest station data from IMS Envista API by date range

        Args:
            station_id (int): IMS Station ID
            from_date (date): From date to query
            to_date (date): to date to query
            channel_id (int): [Optional] Specific Channel Id

        Returns:
            data: Current station meteorological data
        """
        get_url = GET_STATION_DATA_BY_RANGE_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
            str(from_date.strftime("%Y")),
            str(from_date.strftime("%m")),
            str(from_date.strftime("%d")),
            str(to_date.strftime("%Y")),
            str(to_date.strftime("%m")),
            str(to_date.strftime("%d")),
        )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_daily_station_data(
            self, station_id: int, channel_id: int = None
        ) -> StationMeteorologicalReadings:
        """Fetches the daily station data from IMS Envista API

        Args:
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel Id

        Returns:
            data: Current station meteorological data
        """
        get_url = GET_DAILY_STATION_DATA_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
        )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_monthly_station_data(
            self,
            station_id: int,
            channel_id: int = None,
            month: str = None,
            year: str = None,
        ) -> StationMeteorologicalReadings:
        """Fetches monthly station data from IMS Envista API

        Args:
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel Id
            month (str): [Optional] Specific Month in MM format (07)
            year (str):  [Optional] Specific Year in YYYY format (2020)

        Returns:
            data: Current station meteorological data
        """

        if not month or not year:
            get_url = GET_MONTHLY_STATION_DATA_URL.format(
                str(station_id), self._get_channel_id_url_part(channel_id)
            )
        else:
            get_url = GET_MONTHLY_STATION_DATA_BY_MONTH_URL.format(
                str(station_id), self._get_channel_id_url_part(channel_id), year, month
            )
        return station_meteo_data_from_json(self._get_ims_url(get_url))

    def get_all_stations_info(self) -> List[StationInfo]:
        """Fetches all stations data from IMS Envista API

        Returns:
            data: All stations data
        """
        get_url = GET_ALL_STATIONS_DATA_URL
        response = self._get_ims_url(get_url)
        stations = []
        for station in response:
            stations.append(station_from_json(station))
        return stations

    def get_station_info(self, station_id: int) -> StationInfo:
        """Fetches station data from IMS Envista API

        Args:
            station_id (int): IMS Station ID

        Returns:
            data: Current station data
        """
        get_url = GET_SPECIFIC_STATION_DATA_URL.format(str(station_id))
        return station_from_json(self._get_ims_url(get_url))

    def get_all_regions_info(self) -> List[RegionInfo]:
        """Fetches all regions data from IMS Envista API

        Returns:
            data: All stations data
        """
        get_url = GET_ALL_REGIONS_DATA_URL
        response = self._get_ims_url(get_url)
        regions = []
        for region in response:
            stations = []
            for station in region[API_STATIONS]:
                stations.append(station_from_json(station))

            regions.append(
                RegionInfo(region[API_REGION_ID], region[API_NAME], stations)
            )
        return regions

    def get_region_info(self, region_id: int) -> RegionInfo:
        """Fetches region data from IMS Envista API

        Args:
            region_id (int): IMS Region ID

        Returns:
            data: region data
        """
        get_url = GET_SPECIFIC_REGION_DATA_URL.format(str(region_id))
        response = self._get_ims_url(get_url)
        return region_from_json(response)

    def get_metrics_descriptions(self) -> List[IMSVariable]:
class Foo:
    def some_method(self, x, y):
        return x + y;
        """Returns the descriptions of Meteorological Metrics collected by the stations.

        Returns:
            list of IMSVariable, containing description and measuring unit
        """
        return list(VARIABLES.values())


1			"""Module IMSEnvista getting IMS meteorological readings."""
2
3			from __future__ import annotations
4			import json
5			from typing import Optional, List
6
7			from datetime import date
8			import requests
9			from requests.adapters import HTTPAdapter
10			from requests.exceptions import HTTPError
11			from loguru import logger
12			from urllib3 import Retry
13
14			from .const import (
15			GET_LATEST_STATION_DATA_URL,
16			GET_EARLIEST_STATION_DATA_URL,
17			GET_STATION_DATA_BY_DATE_URL,
18			GET_SPECIFIC_STATION_DATA_URL,
19			GET_ALL_STATIONS_DATA_URL,
20			GET_ALL_REGIONS_DATA_URL,
21			API_REGION_ID,
22			API_NAME,
23			API_STATIONS,
24			GET_SPECIFIC_REGION_DATA_URL,
25			GET_DAILY_STATION_DATA_URL,
26			GET_MONTHLY_STATION_DATA_URL,
27			GET_MONTHLY_STATION_DATA_BY_MONTH_URL,
28			GET_STATION_DATA_BY_RANGE_URL,
29			VARIABLES
30			)
31			from .ims_variable import IMSVariable
32			from .meteo_data import (
33			station_meteo_data_from_json,
34			StationMeteorologicalReadings,
35			)
36			from .station_data import StationInfo, station_from_json, region_from_json, RegionInfo
37
38			# ims.gov.il does not support ipv6 yet, `requests` use ipv6 by default
39			# and wait for timeout before trying ipv4, so we have to disable ipv6
40			requests.packages.urllib3.util.connection.HAS_IPV6 = False
41
42			def create_session():
			0 ignored issues – show introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
43			session = requests.Session()
44			retry = Retry(connect=3, backoff_factor=0.5)
45			adapter = HTTPAdapter(max_retries=retry)
46			session.mount('http://', adapter)
47			session.mount('https://', adapter)
48			return session
49
50
51			class IMSEnvista:
52			"""API Wrapper to IMS Envista"""
53
54			def __init__(self, token: str):
55			if not token:
56			raise ValueError
57
58			self.token = token
59			self.session = create_session()
60
61			@staticmethod
62			def _get_channel_id_url_part(channel_id: int) -> str:
63			"""get specific Channel Id url param"""
64			if channel_id:
65			return "/" + str(channel_id)
66			return ""
67
68			def _get_ims_url(self, url: str) -> Optional[dict]:
69			"""Fetches data from IMS url
70
71			Args:
72			url (str): IMS Station ID
73
74			Returns:
75			data: Current station meteorological data
76			"""
77			logger.debug(f"Fetching data from: {url}")
78			try:
79			response = self.session.get(
80			url,
81			headers={
82			"Accept": "application/vnd.github.v3.text-match+json",
83			"Authorization": f"ApiToken {self.token}",
84			},
85			timeout=10,
86			)
87
88			# If the response was successful, no Exception will be raised
89			response.raise_for_status()
90			return json.loads(response.text)
91			except HTTPError as http_err:
92			logger.error(f"HTTP error occurred: {http_err}") # Python 3.6
93			return None
94			except Exception as err:
			0 ignored issues – show Best Practice introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
95			logger.error(f"Other error occurred: {err}") # Python 3.6
96			return None
97
98			def close(self):
99			""" Close Requests Session """
100			self.session.close()
101
102			def get_latest_station_data(
103			self, station_id: int, channel_id: int = None
104			) -> StationMeteorologicalReadings:
105			"""Fetches the latest station data from IMS Envista API
106
107			Args:
108			station_id (int): IMS Station Id
109			channel_id (int): [Optional] Specific Channel Id
110
111			Returns:
112			data: Current station meteorological data
113			"""
114			get_url = GET_LATEST_STATION_DATA_URL.format(
115			str(station_id), self._get_channel_id_url_part(channel_id)
116			)
117			return station_meteo_data_from_json(self._get_ims_url(get_url))
118
119			def get_earliest_station_data(
120			self, station_id: int, channel_id: int = None
121			) -> StationMeteorologicalReadings:
122			"""Fetches the earliest station data from IMS Envista API
123
124			Args:
125			station_id (int): IMS Station ID
126			channel_id (int): [Optional] Specific Channel ID
127
128			Returns:
129			data: Current station meteorological data
130			"""
131			get_url = GET_EARLIEST_STATION_DATA_URL.format(
132			str(station_id), self._get_channel_id_url_part(channel_id)
133			)
134			return station_meteo_data_from_json(self._get_ims_url(get_url))
135
136			def get_station_data_from_date(
137			self, station_id: int, date_to_query: date, channel_id: int = None
138			) -> StationMeteorologicalReadings:
139			"""Fetches latest station data from IMS Envista API by date
140
141			Args:
142			station_id (int): IMS Station ID
143			date_to_query (date): Selected date to query
144			channel_id (int): [Optional] Specific Channel Id
145
146			Returns:
147			data: Current station meteorological data
148			"""
149			get_url = GET_STATION_DATA_BY_DATE_URL.format(
150			str(station_id),
151			self._get_channel_id_url_part(channel_id),
152			str(date_to_query.year),
153			str(date_to_query.month),
154			str(date_to_query.day),
155			)
156			return station_meteo_data_from_json(self._get_ims_url(get_url))
157
158			def get_station_data_by_date_range(
159			self,
160			station_id: int,
161			from_date: date,
162			to_date: date,
163			channel_id: int = None,
164			) -> StationMeteorologicalReadings:
165			"""Fetches latest station data from IMS Envista API by date range
166
167			Args:
168			station_id (int): IMS Station ID
169			from_date (date): From date to query
170			to_date (date): to date to query
171			channel_id (int): [Optional] Specific Channel Id
172
173			Returns:
174			data: Current station meteorological data
175			"""
176			get_url = GET_STATION_DATA_BY_RANGE_URL.format(
177			str(station_id),
178			self._get_channel_id_url_part(channel_id),
179			str(from_date.strftime("%Y")),
180			str(from_date.strftime("%m")),
181			str(from_date.strftime("%d")),
182			str(to_date.strftime("%Y")),
183			str(to_date.strftime("%m")),
184			str(to_date.strftime("%d")),
185			)
186			return station_meteo_data_from_json(self._get_ims_url(get_url))
187
188			def get_daily_station_data(
189			self, station_id: int, channel_id: int = None
190			) -> StationMeteorologicalReadings:
191			"""Fetches the daily station data from IMS Envista API
192
193			Args:
194			station_id (int): IMS Station ID
195			channel_id (int): [Optional] Specific Channel Id
196
197			Returns:
198			data: Current station meteorological data
199			"""
200			get_url = GET_DAILY_STATION_DATA_URL.format(
201			str(station_id),
202			self._get_channel_id_url_part(channel_id),
203			)
204			return station_meteo_data_from_json(self._get_ims_url(get_url))
205
206			def get_monthly_station_data(
207			self,
208			station_id: int,
209			channel_id: int = None,
210			month: str = None,
211			year: str = None,
212			) -> StationMeteorologicalReadings:
213			"""Fetches monthly station data from IMS Envista API
214
215			Args:
216			station_id (int): IMS Station ID
217			channel_id (int): [Optional] Specific Channel Id
218			month (str): [Optional] Specific Month in MM format (07)
219			year (str): [Optional] Specific Year in YYYY format (2020)
220
221			Returns:
222			data: Current station meteorological data
223			"""
224
225			if not month or not year:
226			get_url = GET_MONTHLY_STATION_DATA_URL.format(
227			str(station_id), self._get_channel_id_url_part(channel_id)
228			)
229			else:
230			get_url = GET_MONTHLY_STATION_DATA_BY_MONTH_URL.format(
231			str(station_id), self._get_channel_id_url_part(channel_id), year, month
232			)
233			return station_meteo_data_from_json(self._get_ims_url(get_url))
234
235			def get_all_stations_info(self) -> List[StationInfo]:
236			"""Fetches all stations data from IMS Envista API
237
238			Returns:
239			data: All stations data
240			"""
241			get_url = GET_ALL_STATIONS_DATA_URL
242			response = self._get_ims_url(get_url)
243			stations = []
244			for station in response:
245			stations.append(station_from_json(station))
246			return stations
247
248			def get_station_info(self, station_id: int) -> StationInfo:
249			"""Fetches station data from IMS Envista API
250
251			Args:
252			station_id (int): IMS Station ID
253
254			Returns:
255			data: Current station data
256			"""
257			get_url = GET_SPECIFIC_STATION_DATA_URL.format(str(station_id))
258			return station_from_json(self._get_ims_url(get_url))
259
260			def get_all_regions_info(self) -> List[RegionInfo]:
261			"""Fetches all regions data from IMS Envista API
262
263			Returns:
264			data: All stations data
265			"""
266			get_url = GET_ALL_REGIONS_DATA_URL
267			response = self._get_ims_url(get_url)
268			regions = []
269			for region in response:
270			stations = []
271			for station in region[API_STATIONS]:
272			stations.append(station_from_json(station))
273
274			regions.append(
275			RegionInfo(region[API_REGION_ID], region[API_NAME], stations)
276			)
277			return regions
278
279			def get_region_info(self, region_id: int) -> RegionInfo:
280			"""Fetches region data from IMS Envista API
281
282			Args:
283			region_id (int): IMS Region ID
284
285			Returns:
286			data: region data
287			"""
288			get_url = GET_SPECIFIC_REGION_DATA_URL.format(str(region_id))
289			response = self._get_ims_url(get_url)
290			return region_from_json(response)
291
292			def get_metrics_descriptions(self) -> List[IMSVariable]:
			0 ignored issues – show Coding Style introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
293			"""Returns the descriptions of Meteorological Metrics collected by the stations.
294
295			Returns:
296			list of IMSVariable, containing description and measuring unit
297			"""
298			return list(VARIABLES.values())
299

GuyKh / py-ims-envista

Push — main ( 5bd916...7cb3ea )

ims_envista.ims_envista.IMSEnvista.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ims_envista.ims_envista.IMSEnvista.init() A