forked from ExpDev07/coronavirus-tracker-api
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlocations.py
More file actions
65 lines (55 loc) · 2.17 KB
/
locations.py
File metadata and controls
65 lines (55 loc) · 2.17 KB
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
"""app.router.v2.locations.py"""
from fastapi import HTTPException, Request
from ...enums.sources import Sources
from ...models.location import LocationResponse as Location
from ...models.location import LocationsResponse as Locations
from . import V2
# pylint: disable=unused-argument,too-many-arguments,redefined-builtin
@V2.get("/locations", response_model=Locations, response_model_exclude_unset=True)
async 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 = await 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
if not locations:
raise HTTPException(404, detail=f"Source `{source}` does not have the desired location data.")
# 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],
}
# pylint: disable=invalid-name
@V2.get("/locations/{id}", response_model=Location)
async def get_location_by_id(request: Request, id: int, source: Sources = "jhu", timelines: bool = True):
"""
Getting specific location by id.
"""
location = await request.state.source.get(id)
return {"location": location.serialize(timelines)}