Composition of Locations for Batch Requests

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 = []
-
-    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
+    data = None
 
-        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.getLocations(timelines)
-        if self.latestData:
-            self.previousData = self.latestData
-        self.latestData = {
+        locations = self.getAllLocations(timelines)
+        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 = {
@@ -98,7 +53,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 +78,168 @@ 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 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 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()})
+
+        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
+
+    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"):
+        if Data.__instance != None:
+            raise Exception("Singleton!")
         else:
-            data = self._request("/v2/locations", {"country": country})
-        return data["locations"]
+            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.")
+
+            else:
+                Data.__instance.url = url
+
+            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"]
 
-    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"]
+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."
+
+        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(LocationGroup([Location(country_code="US"), Location(country_code="CA")])))