Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
start = params_dict['start']
interval = params_dict['interval']
# build request URL
if start is None:
timeref = 'current'
else:
if interval is None:
timeref = self._trim_to(
timeformatutils.to_date(start), 'year')
else:
timeref = self._trim_to(
timeformatutils.to_date(start), interval)
fixed_url = '%s/%s,%s/%s.json' % (NO2_INDEX_URL, lat, lon, timeref)
uri = http_client.HttpClient.to_url(fixed_url, self._API_key, None)
_, json_data = self._client.cacheable_get_json(uri)
return json_data
:type zip: string
:param country: the location's country code
:type country: string
:returns: an *Observation* instance or ``None`` if no weather data is
available
:raises: *ParseResponseException* when OWM web API responses' data
cannot be parsed or *APICallException* when OWM web API can not be
reached
"""
stringutils.assert_is_string_or_unicode(zipcode)
stringutils.assert_is_string_or_unicode(country)
encoded_zip = stringutils.encode_to_utf8(zipcode)
encoded_country = stringutils.encode_to_utf8(country)
zip_param = encoded_zip + ',' + encoded_country
params = {'zip': zip_param, 'lang': self._language}
uri = http_client.HttpClient.to_url(OBSERVATION_URL,
self._API_key,
self._subscription_type,
self._use_ssl)
_, json_data = self._wapi.cacheable_get_json(uri, params=params)
return self._parsers['observation'].parse_JSON(json_data)
"precede the current time!")
params['start'] = str(unix_start)
else:
unix_start = None
if end is not None:
unix_end = timeformatutils.to_UNIXtime(end)
params['end'] = str(unix_end)
else:
unix_end = None
if unix_start is not None and unix_end is not None:
if unix_start >= unix_end:
raise ValueError("Error: the start time boundary must "
"precede the end time!")
uri = http_client.HttpClient.to_url(CITY_WEATHER_HISTORY_URL,
self._API_key,
self._subscription_type,
self._use_ssl)
_, json_data = self._wapi.cacheable_get_json(uri, params=params)
return self._parsers['weather_history'].parse_JSON(json_data)
def is_API_online(self):
"""
Returns True if the OWM web API is currently online. A short timeout
is used to determine API service availability.
:returns: bool
"""
params = {'q': 'London,UK'}
uri = http_client.HttpClient.to_url(OBSERVATION_URL,
self._API_key,
self._subscription_type)
try:
_1, _2 = self._wapi.cacheable_get_json(uri, params=params)
return True
except api_call_error.APICallTimeoutError:
return False
def _retrieve_station_history(self, station_ID, limit, interval):
"""
Helper method for station_X_history functions.
"""
params = {'id': station_ID, 'type': interval, 'lang': self._language}
if limit is not None:
params['cnt'] = limit
uri = http_client.HttpClient.to_url(STATION_WEATHER_HISTORY_URL,
self._API_key,
self._subscription_type,
self._use_ssl)
_, json_data = self._wapi.cacheable_get_json(uri, params=params)
station_history = \
self._parsers['station_history'].parse_JSON(json_data)
if station_history is not None:
station_history.set_station_ID(station_ID)
station_history.set_interval(interval)
return station_history
start = params_dict['start']
interval = params_dict['interval']
# build request URL
if start is None:
timeref = 'current'
else:
if interval is None:
timeref = self._trim_to(
timeformatutils.to_date(start), 'year')
else:
timeref = self._trim_to(
timeformatutils.to_date(start), interval)
fixed_url = '%s/%s,%s/%s.json' % (OZONE_URL, lat, lon, timeref)
uri = http_client.HttpClient.to_url(fixed_url, self._API_key, None)
_, json_data = self._client.cacheable_get_json(uri)
return json_data
start = params_dict['start']
interval = params_dict['interval']
# build request URL
if start is None:
timeref = 'current'
else:
if interval is None:
timeref = self._trim_to(
timeformatutils.to_date(start), 'year')
else:
timeref = self._trim_to(
timeformatutils.to_date(start), interval)
fixed_url = '%s/%s,%s/%s.json' % (SO2_INDEX_URL, lat, lon, timeref)
uri = http_client.HttpClient.to_url(fixed_url, self._API_key, None)
_, json_data = self._client.cacheable_get_json(uri)
return json_data
Queries the OWM web API for the currently observed weather at the
specified city ID (eg: 5128581)
:param id: the location's city ID
:type id: int
:returns: an *Observation* instance or ``None`` if no weather data is
available
:raises: *ParseResponseException* when OWM web API responses' data
cannot be parsed or *APICallException* when OWM web API can not be
reached
"""
assert type(id) is int, "'id' must be an int"
if id < 0:
raise ValueError("'id' value must be greater than 0")
params = {'id': id, 'lang': self._language}
uri = http_client.HttpClient.to_url(OBSERVATION_URL,
self._API_key,
self._subscription_type,
self._use_ssl)
_, json_data = self._wapi.cacheable_get_json(uri, params=params)
return self._parsers['observation'].parse_JSON(json_data)
def get_uvi(self, params_dict):
"""
Invokes the UV Index endpoint
:param params_dict: dict of parameters
:returns: a string containing raw JSON data
:raises: *ValueError*, *APICallError*
"""
lat = str(params_dict['lat'])
lon = str(params_dict['lon'])
params = dict(lat=lat, lon=lon)
# build request URL
uri = http_client.HttpClient.to_url(UV_INDEX_URL, self._API_key, None)
_, json_data = self._client.cacheable_get_json(uri, params=params)
return json_data