Creational pattern

app/data/__init__.py

@@ -3,19 +3,26 @@
 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):
+class Source:
+
+    def __init__(self, source):
+        # Mapping of services to data-sources.
+        self.__DATA_SOURCES_LIST = {
+            "jhu": JhuLocationService(),
+            "csbs": CSBSLocationService(),
+            "nyt": NYTLocationService(),
+        }
+
+    def all_data_source(self):
+        return self.__DATA_SOURCES_LIST
+
+    def single_data_source(self, source):
     """
     Retrieves the provided data-source service.
 
     :returns: The service.
     :rtype: LocationService
     """
-    return DATA_SOURCES.get(source.lower())
+    return self.__DATA_SOURCES_LIST.get(source.lower())

app/location/__init__.py

@@ -1,7 +1,7 @@
 """app.location"""
 from ..coordinates import Coordinates
-from ..utils import countries
-from ..utils.populations import country_population
+from ..utils import countries_population
+from ..utils.populations import POPULATION
 
 
 # pylint: disable=redefined-builtin,invalid-name
@@ -11,6 +11,7 @@ class Location:  # pylint: disable=too-many-instance-attributes
     """
 
     def __init__(
+
         self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered,
     ):  # pylint: disable=too-many-arguments
         # General info.

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 Source
 from .routers import V1, V2
 from .utils.httputils import setup_client_session, teardown_client_session
 
@@ -46,6 +46,7 @@
 #######################
 
 # Scout APM
+
 if SETTINGS.scout_name:  # pragma: no cover
     LOGGER.info(f"Adding Scout APM middleware for `{SETTINGS.scout_name}`")
     APP.add_middleware(ScoutMiddleware)
@@ -67,14 +68,14 @@
 )
 APP.add_middleware(GZipMiddleware, minimum_size=1000)
 
-
+SOURCES = Source()
 @APP.middleware("http")
 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 = SOURCE.get_data_source(request.query_params.get("source", default="jhu"))
 
     # Abort with 404 if source cannot be found.
     if not source:

app/routers/v2.py

@@ -3,7 +3,7 @@
 
 from fastapi import APIRouter, HTTPException, Request
 
-from ..data import DATA_SOURCES
+from ..data import Source
 from ..models import LatestResponse, LocationResponse, LocationsResponse
 
 V2 = APIRouter()
@@ -101,10 +101,10 @@ async def get_location_by_id(
     location = await request.state.source.get(id)
     return {"location": location.serialize(timelines)}
 
-
+SOURCE = Source()
 @V2.get("/sources")
 async def sources():
     """
     Retrieves a list of data-sources that are availble to use.
     """
-    return {"sources": list(DATA_SOURCES.keys())}
+    return {"sources": list(SOURCE.all_data_source)}

app/utils/populations.py

@@ -10,51 +10,60 @@
 GEONAMES_URL = "http://api.geonames.org/countryInfoJSON"
 GEONAMES_BACKUP_PATH = "geonames_population_mappings.json"
 
-# Fetching of the populations.
-def fetch_populations(save=False):
-    """
-    Returns a dictionary containing the population of each country fetched from the GeoNames.
-    https://www.geonames.org/
-
-    TODO: only skip writing to the filesystem when deployed with gunicorn, or handle concurent access, or use DB.
-
-    :returns: The mapping of populations.
-    :rtype: dict
-    """
-    LOGGER.info("Fetching populations...")
-
-    # Mapping of populations
-    mappings = {}
-
-    # Fetch the countries.
-    try:
-        countries = requests.get(GEONAMES_URL, params={"username": "dperic"}, timeout=1.25).json()[
-            "geonames"
-        ]
-        # Go through all the countries and perform the mapping.
-        for country in countries:
-            mappings.update({country["countryCode"]: int(country["population"]) or None})
-
-        if mappings and save:
-            LOGGER.info(f"Saving population data to {app.io.save(GEONAMES_BACKUP_PATH, mappings)}")
-    except (json.JSONDecodeError, KeyError, requests.exceptions.Timeout) as err:
-        LOGGER.warning(f"Error pulling population data. {err.__class__.__name__}: {err}")
-        mappings = app.io.load(GEONAMES_BACKUP_PATH)
-        LOGGER.info(f"Using backup data from {GEONAMES_BACKUP_PATH}")
-    # Finally, return the mappings.
-    LOGGER.info("Fetched populations")
-    return mappings
-
-
-# Mapping of alpha-2 codes country codes to population.
-POPULATIONS = fetch_populations()
-
-# Retrieving.
-def country_population(country_code, default=None):
-    """
-    Fetches the population of the country with the provided country code.
-
-    :returns: The population.
-    :rtype: int
-    """
-    return POPULATIONS.get(country_code, default)
+
+class POPULATION:
+    # Fetching of the populations.
+    POPULATION = fetch_population()
+
+    def __init__(self):
+        self.population = POPULATION
+
+    def fetch_populations(save=False):
+        """
+        Returns a dictionary containing the population of each country fetched from the GeoNames.
+        https://www.geonames.org/
+
+        TODO: only skip writing to the filesystem when deployed with gunicorn, or handle concurent access, or use DB.
+
+        :returns: The mapping of populations.
+        :rtype: dict
+        """
+        LOGGER.info("Fetching populations...")
+
+        # Mapping of populations
+        mappings = {}
+
+        # Fetch the countries.
+        try:
+            countries = requests.get(GEONAMES_URL, params={"username": "dperic"}, timeout=1.25).json()[
+                "geonames"
+                ]
+                    # Go through all the countries and perform the mapping.
+            for country in countries:
+                mappings.update({country["countryCode"]: int(country["population"]) or None})
+
+            if mappings and save:
+                LOGGER.info(f"Saving population data to {app.io.save(GEONAMES_BACKUP_PATH, mappings)}")
+                except (json.JSONDecodeError, KeyError, requests.exceptions.Timeout) as err:
+                    LOGGER.warning(f"Error pulling population data. {err.__class__.__name__}: {err}")
+                    mappings = app.io.load(GEONAMES_BACKUP_PATH)
+                    LOGGER.info(f"Using backup data from {GEONAMES_BACKUP_PATH}")
+                    # Finally, return the mappings.
+                    LOGGER.info("Fetched populations")
+                    return mappings
+
+
+    # Mapping of alpha-2 codes country codes to population.
+    POPULATIONS = fetch_populations()
+
+    # Retrieving.
+    def get_country_population(country_code, default=None):
+        """
+        Fetches the population of the country with the provided country code.
+
+        :returns: The population.
+        :rtype: int
+        """
+        return self.population.get(country_code, default)
+
+

tests/test_countries.py

@@ -1,6 +1,6 @@
 import pytest
 
-from app.utils import countries
+from app.utils import countries_population
 
 """
 Todo:
@@ -19,5 +19,20 @@
         ("Others", countries.DEFAULT_COUNTRY_CODE),
     ],
 )
+
+__instance = None
+def getInstance():
+     """ Static access method. """
+    if countries_population.__instance == None:
+        countries_populaiton()
+    return countries_population.__instance
+
+def __init__(self):
+        """ Virtually private constructor. """
+      if countries_population.__instance != None:
+         raise Exception("This class is a singleton!")
+      else:
+         countries_population.__instance = self
+
 def test_countries_country_name__country_code(country_name, expected_country_code):
     assert countries.country_code(country_name) == expected_country_code