ims_envista.ims_envista.IMSEnvista.close() - Code Metrics - Inspection of "Try fix SSLV3 error" - GuyKh/py-ims-envista - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#25)

by Guy

created 2023-09-26 10:02 UTC

ims_envista.ims_envista.IMSEnvista.close() A

↳ Parent: ims_envista.ims_envista.IMSEnvista.get_latest_station_data()

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	3
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 .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

# https://github.com/home-assistant/core/issues/92500#issuecomment-1636743499
ssl_ctx = requests.packages.urllib3.util.ssl_.create_urllib3_context()

ssl_ctx.load_default_certs()
ssl_ctx.options |= 0x4


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

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

        self.token = token

    @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:
            with requests.packages.urllib3.PoolManager(ssl_context=ssl_ctx) as http:

                response = http.request("GET", url, headers={
                    "Accept": "application/vnd.github.v3.text-match+json",
                    "Authorization": f"ApiToken {self.token}"
                }, retries=requests.packages.urllib3.Retry(3), 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 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
			0 ignored issues – show Unused Code introduced 2023-09-26 10:03 UTC by Report Bug Copy Issue Report Unused HTTPAdapter imported from requests.adapters Loading history...
10			from requests.exceptions import HTTPError
11			from loguru import logger
12
13			from .const import (
14			GET_LATEST_STATION_DATA_URL,
15			GET_EARLIEST_STATION_DATA_URL,
16			GET_STATION_DATA_BY_DATE_URL,
17			GET_SPECIFIC_STATION_DATA_URL,
18			GET_ALL_STATIONS_DATA_URL,
19			GET_ALL_REGIONS_DATA_URL,
20			API_REGION_ID,
21			API_NAME,
22			API_STATIONS,
23			GET_SPECIFIC_REGION_DATA_URL,
24			GET_DAILY_STATION_DATA_URL,
25			GET_MONTHLY_STATION_DATA_URL,
26			GET_MONTHLY_STATION_DATA_BY_MONTH_URL,
27			GET_STATION_DATA_BY_RANGE_URL,
28			VARIABLES
29			)
30			from .ims_variable import IMSVariable
31			from .meteo_data import (
32			station_meteo_data_from_json,
33			StationMeteorologicalReadings,
34			)
35			from .station_data import StationInfo, station_from_json, region_from_json, RegionInfo
36
37			# ims.gov.il does not support ipv6 yet, `requests` use ipv6 by default
38			# and wait for timeout before trying ipv4, so we have to disable ipv6
39			requests.packages.urllib3.util.connection.HAS_IPV6 = False
40
41			# https://github.com/home-assistant/core/issues/92500#issuecomment-1636743499
42			ssl_ctx = requests.packages.urllib3.util.ssl_.create_urllib3_context()
			0 ignored issues – show Bug introduced 2023-09-26 10:03 UTC by Report Bug Copy Issue Report The Module `requests.packages` does not seem to have a member named `urllib3`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
43			ssl_ctx.load_default_certs()
44			ssl_ctx.options \|= 0x4
45
46
47			class IMSEnvista:
48			"""API Wrapper to IMS Envista"""
49
50			def __init__(self, token: str):
51			if not token:
52			raise ValueError
53
54			self.token = token
55
56			@staticmethod
57			def _get_channel_id_url_part(channel_id: int) -> str:
58			"""get specific Channel Id url param"""
59			if channel_id:
60			return "/" + str(channel_id)
61			return ""
62
63			def _get_ims_url(self, url: str) -> Optional[dict]:
64			"""Fetches data from IMS url
65
66			Args:
67			url (str): IMS Station ID
68
69			Returns:
70			data: Current station meteorological data
71			"""
72			logger.debug(f"Fetching data from: {url}")
73			try:
74			with requests.packages.urllib3.PoolManager(ssl_context=ssl_ctx) as http:
			0 ignored issues – show Bug introduced 2023-09-26 10:03 UTC by Report Bug Copy Issue Report The Module `requests.packages` does not seem to have a member named `urllib3`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
75			response = http.request("GET", url, headers={
76			"Accept": "application/vnd.github.v3.text-match+json",
77			"Authorization": f"ApiToken {self.token}"
78			}, retries=requests.packages.urllib3.Retry(3), timeout=10)
			0 ignored issues – show Bug introduced 2023-09-26 10:03 UTC by Report Bug Copy Issue Report The Module `requests.packages` does not seem to have a member named `urllib3`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
79
80			# If the response was successful, no Exception will be raised
81			response.raise_for_status()
82			return json.loads(response.text)
83			except HTTPError as http_err:
84			logger.error(f"HTTP error occurred: {http_err}") # Python 3.6
85			return None
86			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...
87			logger.error(f"Other error occurred: {err}") # Python 3.6
88			return None
89
90			def get_latest_station_data(
91			self, station_id: int, channel_id: int = None
92			) -> StationMeteorologicalReadings:
93			"""Fetches the latest station data from IMS Envista API
94
95			Args:
96			station_id (int): IMS Station Id
97			channel_id (int): [Optional] Specific Channel Id
98
99			Returns:
100			data: Current station meteorological data
101			"""
102			get_url = GET_LATEST_STATION_DATA_URL.format(
103			str(station_id), self._get_channel_id_url_part(channel_id)
104			)
105			return station_meteo_data_from_json(self._get_ims_url(get_url))
106
107			def get_earliest_station_data(
108			self, station_id: int, channel_id: int = None
109			) -> StationMeteorologicalReadings:
110			"""Fetches the earliest station data from IMS Envista API
111
112			Args:
113			station_id (int): IMS Station ID
114			channel_id (int): [Optional] Specific Channel ID
115
116			Returns:
117			data: Current station meteorological data
118			"""
119			get_url = GET_EARLIEST_STATION_DATA_URL.format(
120			str(station_id), self._get_channel_id_url_part(channel_id)
121			)
122			return station_meteo_data_from_json(self._get_ims_url(get_url))
123
124			def get_station_data_from_date(
125			self, station_id: int, date_to_query: date, channel_id: int = None
126			) -> StationMeteorologicalReadings:
127			"""Fetches latest station data from IMS Envista API by date
128
129			Args:
130			station_id (int): IMS Station ID
131			date_to_query (date): Selected date to query
132			channel_id (int): [Optional] Specific Channel Id
133
134			Returns:
135			data: Current station meteorological data
136			"""
137			get_url = GET_STATION_DATA_BY_DATE_URL.format(
138			str(station_id),
139			self._get_channel_id_url_part(channel_id),
140			str(date_to_query.year),
141			str(date_to_query.month),
142			str(date_to_query.day),
143			)
144			return station_meteo_data_from_json(self._get_ims_url(get_url))
145
146			def get_station_data_by_date_range(
147			self,
148			station_id: int,
149			from_date: date,
150			to_date: date,
151			channel_id: int = None,
152			) -> StationMeteorologicalReadings:
153			"""Fetches latest station data from IMS Envista API by date range
154
155			Args:
156			station_id (int): IMS Station ID
157			from_date (date): From date to query
158			to_date (date): to date to query
159			channel_id (int): [Optional] Specific Channel Id
160
161			Returns:
162			data: Current station meteorological data
163			"""
164			get_url = GET_STATION_DATA_BY_RANGE_URL.format(
165			str(station_id),
166			self._get_channel_id_url_part(channel_id),
167			str(from_date.strftime("%Y")),
168			str(from_date.strftime("%m")),
169			str(from_date.strftime("%d")),
170			str(to_date.strftime("%Y")),
171			str(to_date.strftime("%m")),
172			str(to_date.strftime("%d")),
173			)
174			return station_meteo_data_from_json(self._get_ims_url(get_url))
175
176			def get_daily_station_data(
177			self, station_id: int, channel_id: int = None
178			) -> StationMeteorologicalReadings:
179			"""Fetches the daily station data from IMS Envista API
180
181			Args:
182			station_id (int): IMS Station ID
183			channel_id (int): [Optional] Specific Channel Id
184
185			Returns:
186			data: Current station meteorological data
187			"""
188			get_url = GET_DAILY_STATION_DATA_URL.format(
189			str(station_id),
190			self._get_channel_id_url_part(channel_id),
191			)
192			return station_meteo_data_from_json(self._get_ims_url(get_url))
193
194			def get_monthly_station_data(
195			self,
196			station_id: int,
197			channel_id: int = None,
198			month: str = None,
199			year: str = None,
200			) -> StationMeteorologicalReadings:
201			"""Fetches monthly station data from IMS Envista API
202
203			Args:
204			station_id (int): IMS Station ID
205			channel_id (int): [Optional] Specific Channel Id
206			month (str): [Optional] Specific Month in MM format (07)
207			year (str): [Optional] Specific Year in YYYY format (2020)
208
209			Returns:
210			data: Current station meteorological data
211			"""
212
213			if not month or not year:
214			get_url = GET_MONTHLY_STATION_DATA_URL.format(
215			str(station_id), self._get_channel_id_url_part(channel_id)
216			)
217			else:
218			get_url = GET_MONTHLY_STATION_DATA_BY_MONTH_URL.format(
219			str(station_id), self._get_channel_id_url_part(channel_id), year, month
220			)
221			return station_meteo_data_from_json(self._get_ims_url(get_url))
222
223			def get_all_stations_info(self) -> List[StationInfo]:
224			"""Fetches all stations data from IMS Envista API
225
226			Returns:
227			data: All stations data
228			"""
229			get_url = GET_ALL_STATIONS_DATA_URL
230			response = self._get_ims_url(get_url)
231			stations = []
232			for station in response:
233			stations.append(station_from_json(station))
234			return stations
235
236			def get_station_info(self, station_id: int) -> StationInfo:
237			"""Fetches station data from IMS Envista API
238
239			Args:
240			station_id (int): IMS Station ID
241
242			Returns:
243			data: Current station data
244			"""
245			get_url = GET_SPECIFIC_STATION_DATA_URL.format(str(station_id))
246			return station_from_json(self._get_ims_url(get_url))
247
248			def get_all_regions_info(self) -> List[RegionInfo]:
249			"""Fetches all regions data from IMS Envista API
250
251			Returns:
252			data: All stations data
253			"""
254			get_url = GET_ALL_REGIONS_DATA_URL
255			response = self._get_ims_url(get_url)
256			regions = []
257			for region in response:
258			stations = []
259			for station in region[API_STATIONS]:
260			stations.append(station_from_json(station))
261
262			regions.append(
263			RegionInfo(region[API_REGION_ID], region[API_NAME], stations)
264			)
265			return regions
266
267			def get_region_info(self, region_id: int) -> RegionInfo:
268			"""Fetches region data from IMS Envista API
269
270			Args:
271			region_id (int): IMS Region ID
272
273			Returns:
274			data: region data
275			"""
276			get_url = GET_SPECIFIC_REGION_DATA_URL.format(str(region_id))
277			response = self._get_ims_url(get_url)
278			return region_from_json(response)
279
280			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...
281			"""Returns the descriptions of Meteorological Metrics collected by the stations.
282
283			Returns:
284			list of IMSVariable, containing description and measuring unit
285			"""
286			return list(VARIABLES.values())
287

GuyKh / py-ims-envista

Pull Request — main (#25)

ims_envista.ims_envista.IMSEnvista.close() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like