Add local JSON as data source

Author: potchangelo

app/data/__init__.py

@@ -1,10 +1,12 @@
 """app.data"""
 from ..services.location.csbs import CSBSLocationService
 from ..services.location.jhu import JhuLocationService
+from ..services.location.localjson import LocalJsonLocationService
 from ..services.location.nyt import NYTLocationService
 
 # Mapping of services to data-sources.
 DATA_SOURCES = {
+    "localjson": LocalJsonLocationService(),
     "jhu": JhuLocationService(),
     "csbs": CSBSLocationService(),
     "nyt": NYTLocationService(),

app/data/locations.json

@@ -74,7 +74,7 @@ async def add_datasource(request: Request, call_next):
     Attach the data source to the request.state.
     """
     # Retrieve the datas ource from query param.
-    source = data_source(request.query_params.get("source", default="jhu"))
+    source = data_source(request.query_params.get("source", default="localjson"))
 
     # Abort with 404 if source cannot be found.
     if not source:

app/main.py

@@ -14,6 +14,7 @@ class Sources(str, enum.Enum):
     A source available for retrieving data.
     """
 
+    LOCALJSON = "localjson"
     JHU = "jhu"
     CSBS = "csbs"
     NYT = "nyt"
@@ -40,11 +41,10 @@ async def get_latest(
 @V2.get("/locations", response_model=LocationsResponse, response_model_exclude_unset=True)
 async def get_locations(
     request: Request,
-    source: Sources = "jhu",
+    source: Sources = Sources.LOCALJSON,
     country_code: str = None,
     province: str = None,
     county: str = None,
-    timelines: bool = False,
 ):
     """
     Getting the locations.
@@ -86,7 +86,7 @@ async def get_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],
+        "locations": [location.serialize() for location in locations],
     }
 
 

app/routers/v2.py

@@ -0,0 +1,105 @@
+"""app.services.location.localjson.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
+
+import app.io
+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
+
+LOGGER = logging.getLogger("services.location.localjson")
+PID = os.getpid()
+
+class LocalJsonLocationService(LocationService):
+    """
+    Service for retrieving locations from local JSON file.
+    """
+
+    async def get_all(self):
+        # Get the locations.
+        locations = await get_locations()
+        return locations
+
+    async def get(self, loc_id):  # pylint: disable=arguments-differ
+        # Get location at the index equal to provided id.
+        locations = await self.get_all()
+
+        return locations[loc_id]
+
+
+# ---------------------------------------------------------------
+
+
+LOCATIONS_PATH = "locations.json"
+
+
+@cached(cache=TTLCache(maxsize=1, ttl=1800))
+async def get_locations():
+    """
+    Retrieves the locations from the categories. The locations are cached for 1 hour.
+
+    :returns: The locations.
+    :rtype: List[Location]
+    """
+
+    data_id = "localjson.locations"
+    LOGGER.info(f"pid:{PID}: {data_id} Requesting data...")
+
+    # Transform JSON to list of object
+    raw_locations: list[dict] = app.io.load(LOCATIONS_PATH)
+    locations: list[TimelinedLocation] = []
+
+    for _, raw_location in enumerate(raw_locations):
+        coordinates = raw_location.get("coordinates", {})
+        timelines = raw_location.get("timelines", {})
+        confirmed_timelines = timelines.get("confirmed", {}).get("timeline", {})
+        deaths_timelines = timelines.get("deaths", {}).get("timeline", {})
+        recovered_timelines = timelines.get("recovered", {}).get("timeline", {})
+
+        locations.append(
+            TimelinedLocation(
+                id=raw_location.get("id"),
+                country=raw_location.get("country"),
+                province=raw_location.get("province"),
+                coordinates=Coordinates(
+                    latitude=coordinates.get("latitude"), 
+                    longitude=coordinates.get("longitude")
+                ),
+                last_updated=datetime.utcnow().isoformat() + "Z",
+                timelines={
+                    "confirmed": Timeline(
+                        timeline={
+                            date: amount
+                            for date, amount in confirmed_timelines.items()
+                        }
+                    ),
+                    "deaths": Timeline(
+                        timeline={
+                            date: amount
+                            for date, amount in deaths_timelines.items()
+                        }
+                    ),
+                    "recovered": Timeline(
+                        timeline={
+                            date: amount
+                            for date, amount in recovered_timelines.items()
+                        }
+                    ),
+                },
+            )
+        )        
+
+    LOGGER.info(f"{data_id} Data normalized")
+
+    return locations