From 8a77d7940bda03484f08e7a2625fd2d6dc9957f9 Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Wed, 17 Mar 2021 18:17:37 -0400 Subject: [PATCH 1/6] Aggregate example 1: Location class added, usage demo in added main() --- COVID19Py/covid19.py | 89 +++++++++++++++++++++++++++++--------------- 1 file changed, 58 insertions(+), 31 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index a68faa7..ebd6700 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -13,7 +13,7 @@ class COVID19(object): mirrors_source = "https://raw.github.com/Kamaropoulos/COVID19Py/master/mirrors.json" mirrors = None - def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source='jhu'): + def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source="jhu"): # Skip mirror checking if custom url was passed if url == self.default_url: # Load mirrors @@ -51,7 +51,7 @@ def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source='jh def _update(self, timelines): latest = self.getLatest() - locations = self.getLocations(timelines) + locations = self.getAllLocations(timelines) if self.latestData: self.previousData = self.latestData self.latestData = { @@ -98,7 +98,7 @@ def getLatest(self) -> List[Dict[str, int]]: data = self._request("/v2/latest") return data["latest"] - def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: + def getAllLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: """ Gets all locations affected by COVID-19, as well as latest case data. :param timelines: Whether timeline information should be returned as well. @@ -123,36 +123,63 @@ def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: return data - def getLocationByCountryCode(self, country_code, timelines=False) -> List[Dict]: + def getLocationData(self, location, timelines=False): """ - :param country_code: String denoting the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + :param location: Location containing any combination of strings denoting... + - the name of the country + - the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + - the country id, an int :param timelines: Whether timeline information should be returned as well. - :return: A list of areas that correspond to the country_code. If the country_code is invalid, it returns an empty list. + :return: Either a list of areas that correspond to the location, or a dictionary with case information for the specified location. """ - data = None - if timelines: - data = self._request("/v2/locations", {"country_code": country_code, "timelines": str(timelines).lower()}) - else: - data = self._request("/v2/locations", {"country_code": country_code}) - return data["locations"] - - def getLocationByCountry(self, country, timelines=False) -> List[Dict]: - """ - :param country: String denoting name of the country - :param timelines: Whether timeline information should be returned as well. - :return: A list of areas that correspond to the country name. If the country is invalid, it returns an empty list. - """ - data = None - if timelines: - data = self._request("/v2/locations", {"country": country, "timelines": str(timelines).lower()}) + # retrieve by country name + if location.country != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country": location.country}) + return data["locations"] + + # retrieve by country code + elif location.country_code != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country_code": location.country_code}) + return data["locations"] + + # retrieve by country id + elif location.country_id != "": + data = self._request("/v2/locations/" + str(location.country_id)) + return data["location"] + + +class Location(object): + country = "" + country_code = "" + country_id = "" + + def __init__(self, country=None, country_code=None, country_id=None): + if country is not None: + self.country = country + if country_code is not None: + self.country_code = country_code + if country_id is not None: + self.country_id = country_id + + def __str__(self): + if self.country == "" and self.country_code == "" and self.country_id == "": + return "Warning! Invalid location provided." else: - data = self._request("/v2/locations", {"country": country}) - return data["locations"] + return "{: <25} ({}) \tid:{}".format(self.country, self.country_code, self.country_id) - def getLocationById(self, country_id: int): - """ - :param country_id: Country Id, an int - :return: A dictionary with case information for the specified location. - """ - data = self._request("/v2/locations/" + str(country_id)) - return data["location"] + +# Testing +def main(): + tracker = COVID19(data_source="jhu") + print(tracker.getLocationData(Location(country_code="CA"))) + +if __name__ == '__main__': + main() \ No newline at end of file From 36fc497270c7c03fdc3831c4eb1f82b848ed8a93 Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Wed, 17 Mar 2021 18:39:38 -0400 Subject: [PATCH 2/6] Aggregate example 1: Location class added, usage demo in added main() --- COVID19Py/covid19.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index ebd6700..4b4475f 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -142,7 +142,7 @@ def getLocationData(self, location, timelines=False): return data["locations"] # retrieve by country code - elif location.country_code != "": + if location.country_code != "": data = None if timelines: data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) @@ -151,7 +151,7 @@ def getLocationData(self, location, timelines=False): return data["locations"] # retrieve by country id - elif location.country_id != "": + if location.country_id != "": data = self._request("/v2/locations/" + str(location.country_id)) return data["location"] @@ -172,8 +172,8 @@ def __init__(self, country=None, country_code=None, country_id=None): def __str__(self): if self.country == "" and self.country_code == "" and self.country_id == "": return "Warning! Invalid location provided." - else: - return "{: <25} ({}) \tid:{}".format(self.country, self.country_code, self.country_id) + + return "{: <25} ({}) \tid:{}".format(self.country, self.country_code, self.country_id) # Testing From 76fb42ed5cdf2ebcf853c484c39f1b0b54f6a717 Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Wed, 17 Mar 2021 18:42:39 -0400 Subject: [PATCH 3/6] Aggregate example 1: Location class added, usage demo added in sample main() --- COVID19Py/covid19.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index 4b4475f..a8fd04a 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -177,9 +177,9 @@ def __str__(self): # Testing -def main(): - tracker = COVID19(data_source="jhu") - print(tracker.getLocationData(Location(country_code="CA"))) +# def main(): +# tracker = COVID19(data_source="jhu") +# print(tracker.getLocationData(Location(country_code="CA"))) -if __name__ == '__main__': - main() \ No newline at end of file +# if __name__ == '__main__': +# main() \ No newline at end of file From 10bfeb0afb9d70bfc24e95adad180af3cb641988 Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Wed, 17 Mar 2021 20:05:28 -0400 Subject: [PATCH 4/6] Aggregate example 2: Data class added, usage demo added in sample main() --- COVID19Py/covid19.py | 127 +++++++++++++++++++++++-------------------- 1 file changed, 68 insertions(+), 59 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index a8fd04a..bd30c4a 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -3,85 +3,40 @@ import json class COVID19(object): - default_url = "https://covid-tracker-us.herokuapp.com" - url = "" - data_source = "" - previousData = None - latestData = None - _valid_data_sources = [] + data = None - mirrors_source = "https://raw.github.com/Kamaropoulos/COVID19Py/master/mirrors.json" - mirrors = None - - def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source="jhu"): - # Skip mirror checking if custom url was passed - if url == self.default_url: - # Load mirrors - response = requests.get(self.mirrors_source) - response.raise_for_status() - self.mirrors = response.json() - - # Try to get sources as a test - for mirror in self.mirrors: - # Set URL of mirror - self.url = mirror["url"] - result = None - try: - result = self._getSources() - except Exception as e: - # URL did not work, reset it and move on - self.url = "" - continue - - # TODO: Should have a better health-check, this is way too hacky... - if "jhu" in result: - # We found a mirror that worked just fine, let's stick with it - break - - # None of the mirrors worked. Raise an error to inform the user. - raise RuntimeError("No available API mirror was found.") - - else: - self.url = url - - self._valid_data_sources = self._getSources() - if data_source not in self._valid_data_sources: - raise ValueError("Invalid data source. Expected one of: %s" % self._valid_data_sources) - self.data_source = data_source + def __init__(self, data=None): + if data is not None: + self.data = data def _update(self, timelines): latest = self.getLatest() locations = self.getAllLocations(timelines) - if self.latestData: - self.previousData = self.latestData - self.latestData = { + if self.data.latestData: + self.data.previousData = self.data.latestData + self.data.latestData = { "latest": latest, "locations": locations } - def _getSources(self): - response = requests.get(self.url + "/v2/sources") - response.raise_for_status() - return response.json()["sources"] - def _request(self, endpoint, params=None): if params is None: params = {} - response = requests.get(self.url + endpoint, {**params, "source":self.data_source}) + response = requests.get(self.data.url + endpoint, {**params, "source":self.data.data_source}) response.raise_for_status() return response.json() def getAll(self, timelines=False): self._update(timelines) - return self.latestData + return self.data.latestData def getLatestChanges(self): changes = None - if self.previousData: + if self.data.previousData: changes = { - "confirmed": self.latestData["latest"]["confirmed"] - self.latestData["latest"]["confirmed"], - "deaths": self.latestData["latest"]["deaths"] - self.latestData["latest"]["deaths"], - "recovered": self.latestData["latest"]["recovered"] - self.latestData["latest"]["recovered"], + "confirmed": self.data.latestData["latest"]["confirmed"] - self.data.latestData["latest"]["confirmed"], + "deaths": self.data.latestData["latest"]["deaths"] - self.data.latestData["latest"]["deaths"], + "recovered": self.data.latestData["latest"]["recovered"] - self.data.latestData["latest"]["recovered"], } else: changes = { @@ -156,6 +111,60 @@ def getLocationData(self, location, timelines=False): return data["location"] +class Data(object): + default_url = "https://covid-tracker-us.herokuapp.com" + url = "" + data_source = "" + previousData = None + latestData = None + _valid_data_sources = [] + + mirrors_source = "https://raw.github.com/Kamaropoulos/COVID19Py/master/mirrors.json" + mirrors = None + + def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source="jhu"): + # Skip mirror checking if custom url was passed + if url == self.default_url: + # Load mirrors + response = requests.get(self.mirrors_source) + response.raise_for_status() + self.mirrors = response.json() + + # Try to get sources as a test + for mirror in self.mirrors: + # Set URL of mirror + self.url = mirror["url"] + result = None + try: + result = self._getSources() + except Exception as e: + # URL did not work, reset it and move on + self.url = "" + continue + + # TODO: Should have a better health-check, this is way too hacky... + if "jhu" in result: + # We found a mirror that worked just fine, let's stick with it + break + + # None of the mirrors worked. Raise an error to inform the user. + raise RuntimeError("No available API mirror was found.") + + else: + self.url = url + + self._valid_data_sources = self._getSources() + if data_source not in self._valid_data_sources: + raise ValueError("Invalid data source. Expected one of: %s" % self._valid_data_sources) + self.data_source = data_source + + def _getSources(self): + response = requests.get(self.url + "/v2/sources") + response.raise_for_status() + return response.json()["sources"] + + + class Location(object): country = "" country_code = "" @@ -178,7 +187,7 @@ def __str__(self): # Testing # def main(): -# tracker = COVID19(data_source="jhu") +# tracker = COVID19(Data()) # print(tracker.getLocationData(Location(country_code="CA"))) # if __name__ == '__main__': From 9623371491ea92b89f0023e78a7c886a4c99de00 Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Tue, 13 Apr 2021 20:49:35 -0400 Subject: [PATCH 5/6] Singleton pattern created for Data class --- COVID19Py/covid19.py | 92 +++++++++++++++++++++++--------------------- 1 file changed, 49 insertions(+), 43 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index bd30c4a..34ad7d9 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -110,8 +110,9 @@ def getLocationData(self, location, timelines=False): data = self._request("/v2/locations/" + str(location.country_id)) return data["location"] +class Data: + __instance = None -class Data(object): default_url = "https://covid-tracker-us.herokuapp.com" url = "" data_source = "" @@ -123,47 +124,56 @@ class Data(object): mirrors = None def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source="jhu"): - # Skip mirror checking if custom url was passed - if url == self.default_url: - # Load mirrors - response = requests.get(self.mirrors_source) - response.raise_for_status() - self.mirrors = response.json() - - # Try to get sources as a test - for mirror in self.mirrors: - # Set URL of mirror - self.url = mirror["url"] - result = None - try: - result = self._getSources() - except Exception as e: - # URL did not work, reset it and move on - self.url = "" - continue - - # TODO: Should have a better health-check, this is way too hacky... - if "jhu" in result: - # We found a mirror that worked just fine, let's stick with it - break - - # None of the mirrors worked. Raise an error to inform the user. - raise RuntimeError("No available API mirror was found.") - + if Data.__instance != None: + raise Exception("Singleton!") else: - self.url = url + Data.__instance = self + + # Skip mirror checking if custom url was passed + if url == Data.__instance.default_url: + # Load mirrors + response = requests.get(Data.__instance.mirrors_source) + response.raise_for_status() + Data.__instance.mirrors = response.json() + + # Try to get sources as a test + for mirror in Data.__instance.mirrors: + # Set URL of mirror + Data.__instance.url = mirror["url"] + result = None + try: + result = Data._getSources() + except Exception as e: + # URL did not work, reset it and move on + Data.__instance.url = "" + continue + + # TODO: Should have a better health-check, this is way too hacky... + if "jhu" in result: + # We found a mirror that worked just fine, let's stick with it + break + + # None of the mirrors worked. Raise an error to inform the user. + raise RuntimeError("No available API mirror was found.") - self._valid_data_sources = self._getSources() - if data_source not in self._valid_data_sources: - raise ValueError("Invalid data source. Expected one of: %s" % self._valid_data_sources) - self.data_source = data_source + else: + Data.__instance.url = url - def _getSources(self): - response = requests.get(self.url + "/v2/sources") - response.raise_for_status() - return response.json()["sources"] + Data.__instance._valid_data_sources = Data._getSources() + if data_source not in Data.__instance._valid_data_sources: + raise ValueError("Invalid data source. Expected one of: %s" % Data.__instance._valid_data_sources) + Data.__instance.data_source = data_source + def Instance(): + if Data.__instance == None: + Data() + return Data.__instance + @staticmethod + def _getSources(): + response = requests.get(Data.__instance.url + "/v2/sources") + response.raise_for_status() + return response.json()["sources"] class Location(object): country = "" @@ -186,9 +196,5 @@ def __str__(self): # Testing -# def main(): -# tracker = COVID19(Data()) -# print(tracker.getLocationData(Location(country_code="CA"))) - -# if __name__ == '__main__': -# main() \ No newline at end of file +if __name__ == '__main__': + print(COVID19(Data()).getLocationData(Location(country_code="CA"))) \ No newline at end of file From 53471fa7cb5486d6d75310a0bedc9d78a5d7df7a Mon Sep 17 00:00:00 2001 From: Nicholas Logan Date: Tue, 13 Apr 2021 21:47:07 -0400 Subject: [PATCH 6/6] Composition of Locations for batch requests --- COVID19Py/covid19.py | 91 +++++++++++++++++++++++++++++++++----------- 1 file changed, 68 insertions(+), 23 deletions(-) diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index 34ad7d9..bec33f3 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -78,7 +78,39 @@ def getAllLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: return data - def getLocationData(self, location, timelines=False): + # def getLocationData(self, location, timelines=False): + # """ + # :param location: Location containing any combination of strings denoting... + # - the name of the country + # - the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + # - the country id, an int + # :param timelines: Whether timeline information should be returned as well. + # :return: Either a list of areas that correspond to the location, or a dictionary with case information for the specified location. + # """ + # # retrieve by country name + # if location.country != "": + # data = None + # if timelines: + # data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) + # else: + # data = self._request("/v2/locations", {"country": location.country}) + # return data["locations"] + + # # retrieve by country code + # if location.country_code != "": + # data = None + # if timelines: + # data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) + # else: + # data = self._request("/v2/locations", {"country_code": location.country_code}) + # return data["locations"] + + # # retrieve by country id + # if location.country_id != "": + # data = self._request("/v2/locations/" + str(location.country_id)) + # return data["location"] + + def getLocationData(self, locationGroup, timelines=False): """ :param location: Location containing any combination of strings denoting... - the name of the country @@ -87,28 +119,34 @@ def getLocationData(self, location, timelines=False): :param timelines: Whether timeline information should be returned as well. :return: Either a list of areas that correspond to the location, or a dictionary with case information for the specified location. """ - # retrieve by country name - if location.country != "": - data = None - if timelines: - data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) - else: - data = self._request("/v2/locations", {"country": location.country}) - return data["locations"] - - # retrieve by country code - if location.country_code != "": - data = None - if timelines: - data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) - else: - data = self._request("/v2/locations", {"country_code": location.country_code}) - return data["locations"] - # retrieve by country id - if location.country_id != "": - data = self._request("/v2/locations/" + str(location.country_id)) - return data["location"] + output = [] + + for location in locationGroup.locList: + # retrieve by country name + if location.country != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country": location.country}) + output += data["locations"] + + # retrieve by country code + elif location.country_code != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country_code": location.country_code}) + output += data["locations"] + + # retrieve by country id + elif location.country_id != "": + data = self._request("/v2/locations/" + str(location.country_id)) + output += data["location"] + + return output class Data: __instance = None @@ -194,7 +232,14 @@ def __str__(self): return "{: <25} ({}) \tid:{}".format(self.country, self.country_code, self.country_id) +class LocationGroup(object): + locList = [] + + def __init__(self, locList=None): + if locList is not None: + self.locList = locList + # Testing if __name__ == '__main__': - print(COVID19(Data()).getLocationData(Location(country_code="CA"))) \ No newline at end of file + print(COVID19(Data()).getLocationData(LocationGroup([Location(country_code="US"), Location(country_code="CA")]))) \ No newline at end of file