Creational design pattern

app/data/__init__.py

@@ -3,19 +3,33 @@
 from ..services.location.jhu import JhuLocationService
 from ..services.location.nyt import NYTLocationService
 
-# Mapping of services to data-sources.
-DATA_SOURCES = {
-    "jhu": JhuLocationService(),
-    "csbs": CSBSLocationService(),
-    "nyt": NYTLocationService(),
-}
-
-
-def data_source(source):
-    """
-    Retrieves the provided data-source service.
-
-    :returns: The service.
-    :rtype: LocationService
-    """
-    return DATA_SOURCES.get(source.lower())
+# Singleton Class
+
+
+class DataSource:
+
+    __instance = None
+
+    __DATA_SOURCES = {
+        "jhu": JhuLocationService(),
+        "csbs": CSBSLocationService(),
+        "nyt": NYTLocationService(),
+    }
+
+    @staticmethod
+    def getInstance():
+        if DataSource.__instance == None:
+            DataSource()
+        return DataSource.__instance
+
+    def __init__(self):
+        if DataSource.__instance != None:
+            raise Exception("This class is a singleton!")
+        else:
+            DataSource.__instance = self
+
+    def setSource(self, source):
+        self.dataSource = self.__DATA_SOURCES.get(source.lower())
+
+    def getSource(self):
+        return self.dataSource

app/main.py

@@ -14,7 +14,7 @@
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 
 from .config import get_settings
-from .data import data_source
+from .data import DataSource, data_source
 from .routers import V1, V2
 from .utils.httputils import setup_client_session, teardown_client_session
 
@@ -80,8 +80,8 @@ async def add_datasource(request: Request, call_next):
     if not source:
         return Response("The provided data-source was not found.", status_code=404)
 
-    # Attach source to request.
-    request.state.source = source
+    # Create Singleton object for source
+    DataSource(source)
 
     # Move on...
     LOGGER.debug(f"source provided: {source.__class__.__name__}")

app/routers/v2.py

@@ -3,7 +3,7 @@
 
 from fastapi import APIRouter, HTTPException, Request
 
-from ..data import DATA_SOURCES
+from data import DataSource
 from ..models import LatestResponse, LocationResponse, LocationsResponse
 
 V2 = APIRouter()
@@ -26,7 +26,7 @@ async def get_latest(
     """
     Getting latest amount of total confirmed cases, deaths, and recoveries.
     """
-    locations = await request.state.source.get_all()
+    locations = await DataSource.getInstance().getSource().get_all()
     return {
         "latest": {
             "confirmed": sum(map(lambda location: location.confirmed, locations)),
@@ -57,7 +57,7 @@ async def get_locations(
     params.pop("timelines", None)
 
     # Retrieve all the locations.
-    locations = await request.state.source.get_all()
+    locations = await DataSource.getInstance().getSource().get_all()
 
     # Attempt to filter out locations with properties matching the provided query params.
     for key, value in params.items():
@@ -98,7 +98,7 @@ async def get_location_by_id(
     """
     Getting specific location by id.
     """
-    location = await request.state.source.get(id)
+    location = await DataSource.getInstance().getSource().get(id)
     return {"location": location.serialize(timelines)}
 
 

app/services/location/nyt.py

@@ -104,18 +104,21 @@ async def get_locations():
             # Make location history for confirmed and deaths from dates.
             # List is tuples of (date, amount) in order of increasing dates.
             confirmed_list = histories["confirmed"]
-            confirmed_history = {date: int(amount or 0) for date, amount in confirmed_list}
+            confirmed_history = {date: int(amount or 0)
+                                 for date, amount in confirmed_list}
 
             deaths_list = histories["deaths"]
-            deaths_history = {date: int(amount or 0) for date, amount in deaths_list}
+            deaths_history = {date: int(amount or 0)
+                              for date, amount in deaths_list}
 
             # 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
+                    # NYT does not provide coordinates
+                    coordinates=Coordinates(None, None),
                     last_updated=datetime.utcnow().isoformat() + "Z",  # since last request
                     timelines={
                         "confirmed": Timeline(