Apply Creational Design Pattern

app/location/locationfactory.py

@@ -0,0 +1,88 @@
+import datetime
+
+from app.coordinates import Coordinates
+from app.location import TimelinedLocation
+from app.location.csbs import CSBSLocation
+from app.location.nyt import NYTLocation
+from app.models import Timeline
+
+
+class LocationFactory:
+    def __init__(self):
+        pass
+
+    @staticmethod
+    def create_location(org_name, params):
+        if org_name == 'NYT':
+            confirmed_history = params["confirmed_history"]
+            deaths_history = params["deaths_history"]
+            return NYTLocation(
+                id=params["index"],
+                state=params["county_state"][1],
+                county=params["county_state"][0],
+                coordinates=Coordinates(None, None),  # NYT does not provide coordinates
+                last_updated=datetime.utcnow().isoformat() + "Z",  # since last request
+                timelines={
+                    "confirmed": Timeline(
+                        timeline={
+                            datetime.strptime(date, "%Y-%m-%d").isoformat() + "Z": amount
+                            for date, amount in confirmed_history.items()
+                        }
+                    ),
+                    "deaths": Timeline(
+                        timeline={
+                            datetime.strptime(date, "%Y-%m-%d").isoformat() + "Z": amount
+                            for date, amount in deaths_history.items()
+                        }
+                    ),
+                    "recovered": Timeline(),
+                },
+            )
+        elif org_name == "JHU":
+            timelines = params["timelines"];
+            return TimelinedLocation(
+                # General info.
+                params["index"],
+                params["country"],
+                params["province"],
+                # Coordinates.
+                Coordinates(latitude=params["coordinates"]["lat"], longitude=params["coordinates"]["long"]),
+                # Last update.
+                datetime.utcnow().isoformat() + "Z",
+                # Timelines (parse dates as ISO).
+                {
+                    "confirmed": Timeline(
+                        timeline={
+                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
+                            for date, amount in timelines["confirmed"].items()
+                        }
+                    ),
+                    "deaths": Timeline(
+                        timeline={
+                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
+                            for date, amount in timelines["deaths"].items()
+                        }
+                    ),
+                    "recovered": Timeline(
+                        timeline={
+                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
+                            for date, amount in timelines["recovered"].items()
+                        }
+                    ),
+                },
+            )
+        elif org_name == "CSBS":
+            item = params["item"]
+            return CSBSLocation(
+                    # General info.
+                    params["index"],
+                    params["state"],
+                    params["county"],
+                    # Coordinates.
+                    Coordinates(item["Latitude"], item["Longitude"]),
+                    # Last update (parse as ISO).
+                    datetime.strptime(params["last_update"], "%Y-%m-%d %H:%M").isoformat() + "Z",
+                    # Statistics.
+                    int(item["Confirmed"] or 0),
+                    int(item["Death"] or 0),
+                )

app/services/location/csbs.py

@@ -8,6 +8,7 @@
 
 from ...caches import check_cache, load_cache
 from ...coordinates import Coordinates
+from ...location import locationfactory
 from ...location.csbs import CSBSLocation
 from ...utils import httputils
 from . import LocationService
@@ -73,23 +74,9 @@ async def get_locations():
 
             # Date string without "EDT" at end.
             last_update = " ".join(item["Last Update"].split(" ")[0:2])
-
+            params = {"index": i, "state": state, "county": county, "item": item, "last_update": last_update}
             # Append to locations.
-            locations.append(
-                CSBSLocation(
-                    # General info.
-                    i,
-                    state,
-                    county,
-                    # Coordinates.
-                    Coordinates(item["Latitude"], item["Longitude"]),
-                    # Last update (parse as ISO).
-                    datetime.strptime(last_update, "%Y-%m-%d %H:%M").isoformat() + "Z",
-                    # Statistics.
-                    int(item["Confirmed"] or 0),
-                    int(item["Death"] or 0),
-                )
-            )
+            locations.append(locationfactory.create_location('CSBS'))
         LOGGER.info(f"{data_id} Data normalized")
         # save the results to distributed cache
         # TODO: fix json serialization

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 TimelinedLocation, locationfactory
 from ...models import Timeline
 from ...utils import countries
 from ...utils import date as date_util
@@ -171,39 +171,10 @@ async def get_locations():
         coordinates = location["coordinates"]
 
         # Create location (supporting timelines) and append.
-        locations.append(
-            TimelinedLocation(
-                # General info.
-                index,
-                location["country"],
-                location["province"],
-                # Coordinates.
-                Coordinates(latitude=coordinates["lat"], longitude=coordinates["long"]),
-                # Last update.
-                datetime.utcnow().isoformat() + "Z",
-                # Timelines (parse dates as ISO).
-                {
-                    "confirmed": Timeline(
-                        timeline={
-                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
-                            for date, amount in timelines["confirmed"].items()
-                        }
-                    ),
-                    "deaths": Timeline(
-                        timeline={
-                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
-                            for date, amount in timelines["deaths"].items()
-                        }
-                    ),
-                    "recovered": Timeline(
-                        timeline={
-                            datetime.strptime(date, "%m/%d/%y").isoformat() + "Z": amount
-                            for date, amount in timelines["recovered"].items()
-                        }
-                    ),
-                },
-            )
-        )
+
+        params= {"index": index, "country": location["country"], "province": location["province"],
+                 "coordinates": coordinates, "timelines": timelines}
+        locations.append(locationfactory.create_location("JHU"), params)
     LOGGER.info(f"{data_id} Data normalized")
 
     # Finally, return the locations.

app/services/location/nyt.py

@@ -1,15 +1,11 @@
 """app.services.location.nyt.py"""
 import csv
 import logging
-from datetime import datetime
-
 from asyncache import cached
 from cachetools import TTLCache
 
 from ...caches import check_cache, load_cache
-from ...coordinates import Coordinates
-from ...location.nyt import NYTLocation
-from ...models import Timeline
+from ...location import locationfactory
 from ...utils import httputils
 from . import LocationService
 
@@ -108,32 +104,10 @@ async def get_locations():
 
             deaths_list = histories["deaths"]
             deaths_history = {date: int(amount or 0) for date, amount in deaths_list}
+            params = {'index': idx, 'county_state': county_state, 'confirmed_history': confirmed_history,
+                      'deaths_history': deaths_history}
+            locations.append(locationfactory.create_location('NYT', params))
 
-            # Normalize the item and append to locations.
-            locations.append(
-                NYTLocation(
-                    id=idx,
-                    state=county_state[1],
-                    county=county_state[0],
-                    coordinates=Coordinates(None, None),  # NYT does not provide coordinates
-                    last_updated=datetime.utcnow().isoformat() + "Z",  # since last request
-                    timelines={
-                        "confirmed": Timeline(
-                            timeline={
-                                datetime.strptime(date, "%Y-%m-%d").isoformat() + "Z": amount
-                                for date, amount in confirmed_history.items()
-                            }
-                        ),
-                        "deaths": Timeline(
-                            timeline={
-                                datetime.strptime(date, "%Y-%m-%d").isoformat() + "Z": amount
-                                for date, amount in deaths_history.items()
-                            }
-                        ),
-                        "recovered": Timeline(),
-                    },
-                )
-            )
         LOGGER.info(f"{data_id} Data normalized")
         # save the results to distributed cache
         # TODO: fix json serialization