Skip to content

Structural pattern #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
M-Antonio12 wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Structural pattern #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
f9fa4c8
This commit contains the application of aggregation. The classes Loca…
M-Antonio12 Jul 24, 2021
bd32166
This commit contains the change to services\location\jhu.py to accoun…
M-Antonio12 Jul 24, 2021
68fda04
This commit applies aggregation within the populations.py.
M-Antonio12 Jul 24, 2021
6edb9cd
In this commit the class name was capitalized and used to refernce in…
M-Antonio12 Jul 24, 2021
d485c94
This commit applies builder design pattern in the router/v2.py for re…
M-Antonio12 Aug 15, 2021
e3851b9
This commit applies faced design pattern in the service/location for …
M-Antonio12 Aug 15, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
235 changes: 118 additions & 117 deletions app/location/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,120 +5,121 @@


# pylint: disable=redefined-builtin,invalid-name
class Location: # pylint: disable=too-many-instance-attributes
"""
A location in the world affected by the coronavirus.
"""

def __init__(
self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered,
): # pylint: disable=too-many-arguments
# General info.
self.id = id
self.country = country.strip()
self.province = province.strip()
self.coordinates = coordinates

# Last update.
self.last_updated = last_updated

# Statistics.
self.confirmed = confirmed
self.deaths = deaths
self.recovered = recovered

@property
def country_code(self):
"""
Gets the alpha-2 code represention of the country. Returns 'XX' if none is found.

:returns: The country code.
:rtype: str
"""
return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper()

@property
def country_population(self):
"""
Gets the population of this location.

:returns: The population.
:rtype: int
"""
return country_population(self.country_code)

def serialize(self):
"""
Serializes the location into a dict.

:returns: The serialized location.
:rtype: dict
"""
return {
# General info.
"id": self.id,
"country": self.country,
"country_code": self.country_code,
"country_population": self.country_population,
"province": self.province,
# Coordinates.
"coordinates": self.coordinates.serialize(),
# Last updated.
"last_updated": self.last_updated,
# Latest data (statistics).
"latest": {
"confirmed": self.confirmed,
"deaths": self.deaths,
"recovered": self.recovered,
},
}


class TimelinedLocation(Location):
"""
A location with timelines.
"""

# pylint: disable=too-many-arguments
def __init__(self, id, country, province, coordinates, last_updated, timelines):
super().__init__(
# General info.
id,
country,
province,
coordinates,
last_updated,
# Statistics (retrieve latest from timelines).
confirmed=timelines.get("confirmed").latest or 0,
deaths=timelines.get("deaths").latest or 0,
recovered=timelines.get("recovered").latest or 0,
)

# Set timelines.
self.timelines = timelines

# pylint: disable=arguments-differ
def serialize(self, timelines=False):
"""
Serializes the location into a dict.

:param timelines: Whether to include the timelines.
:returns: The serialized location.
:rtype: dict
"""
serialized = super().serialize()

# Whether to include the timelines or not.
if timelines:
serialized.update(
{
"timelines": {
# Serialize all the timelines.
key: value.serialize()
for (key, value) in self.timelines.items()
}
}
)

# Return the serialized location.
return serialized
class Locations:
class Location: # pylint: disable=too-many-instance-attributes
"""
A location in the world affected by the coronavirus.
"""

def __init__(
self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered,
): # pylint: disable=too-many-arguments
# General info.
self.id = id
self.country = country.strip()
self.province = province.strip()
self.coordinates = coordinates

# Last update.
self.last_updated = last_updated

# Statistics.
self.confirmed = confirmed
self.deaths = deaths
self.recovered = recovered

@property
def country_code(self):
"""
Gets the alpha-2 code represention of the country. Returns 'XX' if none is found.

:returns: The country code.
:rtype: str
"""
return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper()

@property
def country_population(self):
"""
Gets the population of this location.

:returns: The population.
:rtype: int
"""
return country_population(self.country_code)

def serialize(self):
"""
Serializes the location into a dict.

:returns: The serialized location.
:rtype: dict
"""
return {
# General info.
"id": self.id,
"country": self.country,
"country_code": self.country_code,
"country_population": self.country_population,
"province": self.province,
# Coordinates.
"coordinates": self.coordinates.serialize(),
# Last updated.
"last_updated": self.last_updated,
# Latest data (statistics).
"latest": {
"confirmed": self.confirmed,
"deaths": self.deaths,
"recovered": self.recovered,
},
}


class TimelinedLocation(Location):
"""
A location with timelines.
"""

# pylint: disable=too-many-arguments
def __init__(self, id, country, province, coordinates, last_updated, timelines):
super().__init__(
# General info.
id,
country,
province,
coordinates,
last_updated,
# Statistics (retrieve latest from timelines).
confirmed=timelines.get("confirmed").latest or 0,
deaths=timelines.get("deaths").latest or 0,
recovered=timelines.get("recovered").latest or 0,
)

# Set timelines.
self.timelines = timelines

# pylint: disable=arguments-differ
def serialize(self, timelines=False):
"""
Serializes the location into a dict.

:param timelines: Whether to include the timelines.
:returns: The serialized location.
:rtype: dict
"""
serialized = super().serialize()

# Whether to include the timelines or not.
if timelines:
serialized.update(
{
"timelines": {
# Serialize all the timelines.
key: value.serialize()
for (key, value) in self.timelines.items()
}
}
)

# Return the serialized location.
return serialized
3 changes: 2 additions & 1 deletion app/routers/v1.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
"""app.routers.v1.py"""
from fastapi import APIRouter

from ..services.location.jhu import get_category

from ..services.location.jhu_facade import get_category

V1 = APIRouter()

Expand Down
43 changes: 31 additions & 12 deletions app/routers/v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""app.routers.v2"""
from abc import ABCMeta, abstractclassmethod
import enum

from fastapi import APIRouter, HTTPException, Request
Expand All @@ -18,6 +19,33 @@ class Sources(str, enum.Enum):
CSBS = "csbs"
NYT = "nyt"

class IBuilder(metaclass=ABCMeta):
@staticmethod
@abstractclassmethod

def buildLatest():
"Display latest confirmed, deaths and recoverd"
def buildLocation():
"Display location"


class Builder(IBuilder):
async def buildLatest(request: Request, source: Sources = Sources.JHU):
locations = await 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)),
}
}

async def buildLocation( request: Request, timelines):
locations = await request.state.source.get_all()
return {
"locations": [location.serialize(timelines) for location in locations],
}


@V2.get("/latest", response_model=LatestResponse)
async def get_latest(
Expand All @@ -26,13 +54,9 @@ async def get_latest(
"""
Getting latest amount of total confirmed cases, deaths, and recoveries.
"""
locations = await 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)),
}
Builder.buildLatest()
}


Expand Down Expand Up @@ -81,12 +105,7 @@ async def get_locations(

# 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],
Builder.buildLatest, Builder.buildLocation( request, timelines)
}


Expand Down
5 changes: 3 additions & 2 deletions app/services/location/jhu.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

from ...caches import check_cache, load_cache
from ...coordinates import Coordinates
from ...location import TimelinedLocation
from ...location import Locations
from ...models import Timeline
from ...utils import countries
from ...utils import date as date_util
Expand Down Expand Up @@ -169,10 +169,11 @@ async def get_locations():

# Grab coordinates.
coordinates = location["coordinates"]
locationTimeLined = Locations.TimelinedLocation

# Create location (supporting timelines) and append.
locations.append(
TimelinedLocation(
locationTimeLined(
# General info.
index,
location["country"],
Expand Down
26 changes: 26 additions & 0 deletions app/services/location/jhu_facade.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
"""app.services.location.jhu.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

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
import jhu

class facade():
jhu.JhuLocationService.get()
jhu.JhuLocationService.get_all()
jhu.get_category()
jhu.get_locations()
jhu.parse_history()
Loading