move routes to new files

Author: ExpDev07

app/enums/sources.py

@@ -0,0 +1,8 @@
+from enum import Enum
+
+class Sources(str, Enum):
+    """
+    A source available for retrieving data.
+    """
+    jhu = 'jhu'
+    csbs = 'csbs'

app/main.py

@@ -6,34 +6,22 @@
 import logging
 import os
 import reprlib
-from typing import Dict, List
 
 import fastapi
 import pydantic
 import uvicorn
 
+from typing import Dict, List
+
 from fastapi.middleware.wsgi import WSGIMiddleware
 from fastapi.middleware.cors import CORSMiddleware
 
 from .core import create_app
-from .data import data_source, data_sources
+from .data import data_source
 
 from .models.location import LocationResponse as Location, LocationsResponse as Locations
 from .models.latest import LatestResponse as Latest
 
-# ################
-# Dependencies
-# ################
-
-
-class Sources(str, enum.Enum):
-    """
-    A source available for retrieving data.
-    """
-    jhu = 'jhu'
-    csbs = 'csbs'
-
-
 # ############
 # FastAPI App
 # ############
@@ -98,94 +86,13 @@ async def handle_validation_error(
 
 
 # ################
-# Routes
+# Routing
 # ################
 
-V2 = fastapi.APIRouter()
-
-@V2.get('/latest', response_model=Latest)
-def get_latest(request: fastapi.Request, source: Sources = 'jhu'):
-    """
-    Getting latest amount of total confirmed cases, deaths, and recoveries.
-    """
-    locations = 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)),
-        }
-    }
-
-
-@V2.get(
-    '/locations', response_model=Locations, response_model_exclude_unset=True
-)
-def get_locations(
-    request: fastapi.Request,
-    source: Sources = 'jhu',
-    country_code: str = None,
-    province: str = None,
-    county: str = None,
-    timelines: bool = False,
-):
-    """
-    Getting the locations.
-    """
-    # All query paramameters.
-    params = dict(request.query_params)
-
-    # Remove reserved params.
-    params.pop('source', None)
-    params.pop('timelines', None)
-
-    # Retrieve all the locations.
-    locations = request.state.source.get_all()
-
-    # Attempt to filter out locations with properties matching the provided query params.
-    for key, value in params.items():
-        # Clean keys for security purposes.
-        key   = key.lower()
-        value = value.lower().strip('__')
-
-        # Do filtering.
-        try:
-            locations = [location for location in locations if str(getattr(location, key)).lower() == str(value)]
-        except AttributeError:
-            pass
-
-    # 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],
-    }
-
-
-@V2.get('/locations/{id}', response_model=Location)
-def get_location_by_id(request: fastapi.Request, id: int, source: Sources = 'jhu', timelines: bool = True):
-    """
-    Getting specific location by id.
-    """
-    return {
-        'location': request.state.source.get(id).serialize(timelines)
-    }
-
-
-@V2.get('/sources')
-async def sources():
-    """
-    Retrieves a list of data-sources that are availble to use.
-    """
-    return {
-        'sources': list(data_sources.keys())
-    }
+from .router import router
 
 # Include routers.
-APP.include_router(V2, prefix='/v2', tags=['v2'])
+APP.include_router(router, prefix='/v2', tags=['v2'])
 
 # mount the existing Flask app
 # v1 @ /

app/router/__init__.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+# Create the router.
+router = APIRouter()
+
+# The routes.
+from . import latest, sources, locations

app/router/latest.py

@@ -0,0 +1,18 @@
+from fastapi import Request
+from . import router
+from ..enums.sources import Sources
+from ..models.latest import LatestResponse as Latest
+
+@router.get('/latest', response_model=Latest)
+def get_latest(request: Request, source: Sources = 'jhu'):
+    """
+    Getting latest amount of total confirmed cases, deaths, and recoveries.
+    """
+    locations = 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)),
+        }
+    }

app/router/locations.py

@@ -0,0 +1,60 @@
+from fastapi import Request
+from . import router
+from ..enums.sources import Sources
+from ..models.location import LocationResponse as Location, LocationsResponse as Locations
+
+@router.get(
+    '/locations', response_model=Locations, response_model_exclude_unset=True
+)
+def get_locations(
+    request: Request,
+    source: Sources = 'jhu',
+    country_code: str = None,
+    province: str = None,
+    county: str = None,
+    timelines: bool = False,
+):
+    """
+    Getting the locations.
+    """
+    # All query paramameters.
+    params = dict(request.query_params)
+
+    # Remove reserved params.
+    params.pop('source', None)
+    params.pop('timelines', None)
+
+    # Retrieve all the locations.
+    locations = request.state.source.get_all()
+
+    # Attempt to filter out locations with properties matching the provided query params.
+    for key, value in params.items():
+        # Clean keys for security purposes.
+        key   = key.lower()
+        value = value.lower().strip('__')
+
+        # Do filtering.
+        try:
+            locations = [location for location in locations if str(getattr(location, key)).lower() == str(value)]
+        except AttributeError:
+            pass
+
+    # 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],
+    }
+
+
+@router.get('/locations/{id}', response_model=Location)
+def get_location_by_id(request: Request, id: int, source: Sources = 'jhu', timelines: bool = True):
+    """
+    Getting specific location by id.
+    """
+    return {
+        'location': request.state.source.get(id).serialize(timelines)
+    }

app/router/sources.py

@@ -0,0 +1,11 @@
+from . import router
+from ..data import data_sources
+
+@router.get('/sources')
+async def sources():
+    """
+    Retrieves a list of data-sources that are availble to use.
+    """
+    return {
+        'sources': list(data_sources.keys())
+    }