coronavirus-tracker-api/app/models.py at observer-pattern · anjilag/coronavirus-tracker-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
"""app.models.py"""
from typing import Dict, List

from pydantic import BaseModel, validator
from abc import ABCMeta, abstractmethod

class Observer(metaclass=ABCMeta):
    @abstractmethod
    def update(self, confirmed, deaths, recovered): pass

class Subject(metaclass=ABCMeta):
    @abstractmethod
    def subscribe(self, obs): pass

    @abstractmethod
    def unsubscribe(self, obs): pass

    @abstractmethod
    def notify(self, obs): pass

class Latest(BaseModel, Subject):
    """
    Latest model.
    """

    confirmed: int
    deaths: int
    recovered: int
    subscribers = []

    def subscribe(self, obs):
        self.subscribers.append(obs)

    def unsubscribe(self, obs):
        self.subscribers.remove(obs)

    def notify(self):
        for sub in self.subscribers:
            sub.update(self.confirmed, self.deaths, self.recovered)

class LatestResponse(BaseModel, Observer):
    """
    Response for latest.
    """

    confirmed: int
    deaths:int
    recovered:int

    def update(self, confirmed, deaths, recovered):
        self.confirmed = confirmed
        self.deaths = deaths
        self.recovered = recovered


class Timeline(BaseModel):
    """
    Timeline model.
    """

    timeline: Dict[str, int] = {}

    @validator("timeline")
    @classmethod
    def sort_timeline(cls, value):
        """Sort the timeline history before inserting into the model"""
        return dict(sorted(value.items()))

    @property
    def latest(self):
        """Get latest available history value."""
        return list(self.timeline.values())[-1] if self.timeline else 0

    def serialize(self):
        """
        Serialize the model into dict
        TODO: override dict() instead of using serialize
        """
        return {**self.dict(), "latest": self.latest}


class Timelines(BaseModel):
    """
    Timelines model.
    """

    confirmed: Timeline
    deaths: Timeline
    recovered: Timeline


class Location(BaseModel, Observer):
    """
    Location model.
    """

    id: int
    country: str
    country_code: str
    country_population: int = None
    province: str = ""
    county: str = ""
    last_updated: str  # TODO use datetime.datetime type.
    coordinates: Dict
    confirmed: int
    deaths:int
    recovered:int
    timelines: Timelines = {}

    def update(self, confirmed, deaths, recovered):
        self.confirmed = confirmed
        self.deaths = deaths
        self.recovered = recovered

class LocationResponse(BaseModel):
    """
    Response for location.
    """

    location: Location


class LocationsResponse(BaseModel):
    """
    Response for locations.
    """

    confirmed:int
    deaths:int
    recovered:int
    locations: List[Location] = []

    def update(self, confirmed, deaths, recovered):
        self.confirmed = confirmed
        self.deaths = deaths
        self.recovered = recovered

    locations: List[Location] = []