State pattern

app/location/__init__.py

@@ -5,120 +5,121 @@
 
 
 # pylint: disable=redefined-builtin,invalid-name
-class Location:  # pylint: disable=too-many-instance-attributes
-    """
-    A location in the world affected by the coronavirus.
-    """
-
-    def __init__(
-        self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered,
-    ):  # pylint: disable=too-many-arguments
-        # General info.
-        self.id = id
-        self.country = country.strip()
-        self.province = province.strip()
-        self.coordinates = coordinates
-
-        # Last update.
-        self.last_updated = last_updated
-
-        # Statistics.
-        self.confirmed = confirmed
-        self.deaths = deaths
-        self.recovered = recovered
-
-    @property
-    def country_code(self):
-        """
-        Gets the alpha-2 code represention of the country. Returns 'XX' if none is found.
-
-        :returns: The country code.
-        :rtype: str
-        """
-        return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper()
-
-    @property
-    def country_population(self):
-        """
-        Gets the population of this location.
-
-        :returns: The population.
-        :rtype: int
-        """
-        return country_population(self.country_code)
-
-    def serialize(self):
-        """
-        Serializes the location into a dict.
-
-        :returns: The serialized location.
-        :rtype: dict
-        """
-        return {
-            # General info.
-            "id": self.id,
-            "country": self.country,
-            "country_code": self.country_code,
-            "country_population": self.country_population,
-            "province": self.province,
-            # Coordinates.
-            "coordinates": self.coordinates.serialize(),
-            # Last updated.
-            "last_updated": self.last_updated,
-            # Latest data (statistics).
-            "latest": {
-                "confirmed": self.confirmed,
-                "deaths": self.deaths,
-                "recovered": self.recovered,
-            },
-        }
-
-
-class TimelinedLocation(Location):
-    """
-    A location with timelines.
-    """
-
-    # pylint: disable=too-many-arguments
-    def __init__(self, id, country, province, coordinates, last_updated, timelines):
-        super().__init__(
-            # General info.
-            id,
-            country,
-            province,
-            coordinates,
-            last_updated,
-            # Statistics (retrieve latest from timelines).
-            confirmed=timelines.get("confirmed").latest or 0,
-            deaths=timelines.get("deaths").latest or 0,
-            recovered=timelines.get("recovered").latest or 0,
-        )
-
-        # Set timelines.
-        self.timelines = timelines
-
-    # pylint: disable=arguments-differ
-    def serialize(self, timelines=False):
-        """
-        Serializes the location into a dict.
-
-        :param timelines: Whether to include the timelines.
-        :returns: The serialized location.
-        :rtype: dict
-        """
-        serialized = super().serialize()
-
-        # Whether to include the timelines or not.
-        if timelines:
-            serialized.update(
-                {
-                    "timelines": {
-                        # Serialize all the timelines.
-                        key: value.serialize()
-                        for (key, value) in self.timelines.items()
-                    }
-                }
-            )
-
-        # Return the serialized location.
-        return serialized
+class Locations:
+  class Location:  # pylint: disable=too-many-instance-attributes
+      """
+      A location in the world affected by the coronavirus.
+      """
+
+      def __init__(
+          self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered,
+      ):  # pylint: disable=too-many-arguments
+          # General info.
+          self.id = id
+          self.country = country.strip()
+          self.province = province.strip()
+          self.coordinates = coordinates
+
+          # Last update.
+          self.last_updated = last_updated
+
+          # Statistics.
+          self.confirmed = confirmed
+          self.deaths = deaths
+          self.recovered = recovered
+
+      @property
+      def country_code(self):
+          """
+          Gets the alpha-2 code represention of the country. Returns 'XX' if none is found.
+
+          :returns: The country code.
+          :rtype: str
+          """
+          return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper()
+
+      @property
+      def country_population(self):
+          """
+          Gets the population of this location.
+
+          :returns: The population.
+          :rtype: int
+          """
+          return country_population(self.country_code)
+
+      def serialize(self):
+          """
+          Serializes the location into a dict.
+
+          :returns: The serialized location.
+          :rtype: dict
+          """
+          return {
+              # General info.
+              "id": self.id,
+              "country": self.country,
+              "country_code": self.country_code,
+              "country_population": self.country_population,
+              "province": self.province,
+              # Coordinates.
+              "coordinates": self.coordinates.serialize(),
+              # Last updated.
+              "last_updated": self.last_updated,
+              # Latest data (statistics).
+              "latest": {
+                  "confirmed": self.confirmed,
+                  "deaths": self.deaths,
+                  "recovered": self.recovered,
+              },
+          }
+
+
+  class TimelinedLocation(Location):
+      """
+      A location with timelines.
+      """
+
+      # pylint: disable=too-many-arguments
+      def __init__(self, id, country, province, coordinates, last_updated, timelines):
+          super().__init__(
+              # General info.
+              id,
+              country,
+              province,
+              coordinates,
+              last_updated,
+              # Statistics (retrieve latest from timelines).
+              confirmed=timelines.get("confirmed").latest or 0,
+              deaths=timelines.get("deaths").latest or 0,
+              recovered=timelines.get("recovered").latest or 0,
+          )
+
+          # Set timelines.
+          self.timelines = timelines
+
+      # pylint: disable=arguments-differ
+      def serialize(self, timelines=False):
+          """
+          Serializes the location into a dict.
+
+          :param timelines: Whether to include the timelines.
+          :returns: The serialized location.
+          :rtype: dict
+          """
+          serialized = super().serialize()
+
+          # Whether to include the timelines or not.
+          if timelines:
+              serialized.update(
+                  {
+                      "timelines": {
+                          # Serialize all the timelines.
+                          key: value.serialize()
+                          for (key, value) in self.timelines.items()
+                      }
+                  }
+              )
+
+          # Return the serialized location.
+          return serialized

app/main.py

@@ -13,6 +13,8 @@
 from scout_apm.async_.starlette import ScoutMiddleware
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 
+from app.utils import httputils
+
 from .config import get_settings
 from .data import data_source
 from .routers import V1, V2
@@ -112,7 +114,15 @@ async def handle_validation_error(
 # Include routers.
 APP.include_router(V1, prefix="", tags=["v1"])
 APP.include_router(V2, prefix="/v2", tags=["v2"])
-
+async def shutdown():
+    timer = 5000
+    state = "on"
+    timer = timer - 1
+    if(timer == 0):
+       state = "idle"
+
+    if(state == "idle"):
+       httputils.teardown_client_session() 
 
 # Running of app.
 if __name__ == "__main__":

app/routers/v1.py

@@ -1,7 +1,8 @@
 """app.routers.v1.py"""
 from fastapi import APIRouter
 
-from ..services.location.jhu import get_category
+
+from ..services.location.jhu_facade import get_category
 
 V1 = APIRouter()
 

app/routers/v2.py

@@ -1,4 +1,5 @@
 """app.routers.v2"""
+from abc import ABCMeta, abstractclassmethod
 import enum
 
 from fastapi import APIRouter, HTTPException, Request
@@ -18,6 +19,33 @@ class Sources(str, enum.Enum):
     CSBS = "csbs"
     NYT = "nyt"
 
+class IBuilder(metaclass=ABCMeta):
+    @staticmethod
+    @abstractclassmethod
+
+    def buildLatest():
+        "Display latest confirmed, deaths and recoverd"
+    def buildLocation():
+        "Display location"
+
+
+class Builder(IBuilder):
+   async def buildLatest(request: Request, source: Sources = Sources.JHU):
+        locations = await request.state.source.get_all()
+        return{
+          "latest": {
+            "confirmed": sum(map(lambda location: location.confirmed, locations)),
+            "deaths": sum(map(lambda location: location.deaths, locations)),
+            "recovered": sum(map(lambda location: location.recovered, locations)),
+          }
+        }
+
+   async def buildLocation( request: Request, timelines):
+        locations = await request.state.source.get_all()
+        return {
+             "locations": [location.serialize(timelines) for location in locations],
+        }
+
 
 @V2.get("/latest", response_model=LatestResponse)
 async def get_latest(
@@ -26,13 +54,9 @@ async def get_latest(
     """
     Getting latest amount of total confirmed cases, deaths, and recoveries.
     """
-    locations = await request.state.source.get_all()
+
     return {
-        "latest": {
-            "confirmed": sum(map(lambda location: location.confirmed, locations)),
-            "deaths": sum(map(lambda location: location.deaths, locations)),
-            "recovered": sum(map(lambda location: location.recovered, locations)),
-        }
+        Builder.buildLatest()
     }
 
 
@@ -81,12 +105,7 @@ async def get_locations(
 
     # Return final serialized data.
     return {
-        "latest": {
-            "confirmed": sum(map(lambda location: location.confirmed, locations)),
-            "deaths": sum(map(lambda location: location.deaths, locations)),
-            "recovered": sum(map(lambda location: location.recovered, locations)),
-        },
-        "locations": [location.serialize(timelines) for location in locations],
+        Builder.buildLatest, Builder.buildLocation( request, timelines)
     }
 
 

app/services/location/jhu.py

@@ -10,7 +10,7 @@
 
 from ...caches import check_cache, load_cache
 from ...coordinates import Coordinates
-from ...location import TimelinedLocation
+from ...location import Locations
 from ...models import Timeline
 from ...utils import countries
 from ...utils import date as date_util
@@ -169,10 +169,11 @@ async def get_locations():
 
         # Grab coordinates.
         coordinates = location["coordinates"]
+        locationTimeLined = Locations.TimelinedLocation
 
         # Create location (supporting timelines) and append.
         locations.append(
-            TimelinedLocation(
+            locationTimeLined(
                 # General info.
                 index,
                 location["country"],

app/services/location/jhu_facade.py

@@ -0,0 +1,26 @@
+"""app.services.location.jhu.py"""
+import csv
+import logging
+import os
+from datetime import datetime
+from pprint import pformat as pf
+
+from asyncache import cached
+from cachetools import TTLCache
+
+from ...caches import check_cache, load_cache
+from ...coordinates import Coordinates
+from ...location import TimelinedLocation
+from ...models import Timeline
+from ...utils import countries
+from ...utils import date as date_util
+from ...utils import httputils
+from . import LocationService
+import jhu
+
+class facade():
+    jhu.JhuLocationService.get()
+    jhu.JhuLocationService.get_all()
+    jhu.get_category()
+    jhu.get_locations()
+    jhu.parse_history()