新冠病毒疫情跟踪器应用后台接口开发完成

Author: liaogx

coronavirus/main.py

@@ -4,18 +4,18 @@
 
 from typing import List
 
-from fastapi import APIRouter, Depends, HTTPException
-from fastapi.staticfiles import StaticFiles
+import requests
+from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Request
 from fastapi.templating import Jinja2Templates
+from pydantic import HttpUrl
 from sqlalchemy.orm import Session
 
 from coronavirus import crud, schemas
 from coronavirus.database import engine, Base, SessionLocal
+from coronavirus.models import City, Data
 
 application = APIRouter()
 
-# mount表示将某个目录下一个完全独立的应用挂载过来，这个不会在API交互文档中显示
-application.mount('/static', StaticFiles(directory='./coronavirus/static'), name='static')
 templates = Jinja2Templates(directory='./coronavirus/templates')
 
 Base.metadata.create_all(bind=engine)
@@ -58,7 +58,57 @@ def create_data_for_city(city: str, data: schemas.CreateData, db: Session = Depe
     return data
 
 
-@application.get("/get_data", response_model=List[schemas.ReadData])
-def get_data(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
-    data = crud.get_data(db, skip=skip, limit=limit)
+@application.get("/get_data")
+def get_data(city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    data = crud.get_data(db, city=city, skip=skip, limit=limit)
     return data
+
+
+def bg_task(url: HttpUrl, db: Session):
+    """这里注意一个坑，不要在后台任务的参数中db: Session = Depends(get_db)这样导入依赖"""
+
+    city_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=false")
+
+    if 200 == city_data.status_code:
+        db.query(City).delete()  # 同步数据前先清空原有的数据
+        for location in city_data.json()["locations"]:
+            city = {
+                "province": location["province"],
+                "country": location["country"],
+                "country_code": "CN",
+                "country_population": location["country_population"]
+            }
+            crud.create_city(db=db, city=schemas.CreateCity(**city))
+
+    coronavirus_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=true")
+
+    if 200 == coronavirus_data.status_code:
+        db.query(Data).delete()
+        for city in coronavirus_data.json()["locations"]:
+            db_city = crud.get_city_by_name(db=db, name=city["province"])
+            for date, confirmed in city["timelines"]["confirmed"]["timeline"].items():
+                data = {
+                    "date": date.split("T")[0],  # 把'2020-12-31T00:00:00Z' 变成 ‘2020-12-31’
+                    "confirmed": confirmed,
+                    "deaths": city["timelines"]["deaths"]["timeline"][date],
+                    "recovered": 0  # 每个城市每天有多少人痊愈，这种数据没有
+                }
+                # 这个city_id是city表中的主键ID，不是coronavirus_data数据里的ID
+                crud.create_city_data(db=db, data=schemas.CreateData(**data), city_id=db_city.id)
+
+
+@application.get("/sync_coronavirus_data/jhu")
+def sync_coronavirus_data(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
+    """从Johns Hopkins University同步COVID-19数据"""
+    background_tasks.add_task(bg_task, "https://coronavirus-tracker-api.herokuapp.com/v2/locations", db)
+    return {"message": "正在后台同步数据..."}
+
+
+@application.get("/")
+def coronavirus(request: Request, city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
+    data = crud.get_data(db, city=city, skip=skip, limit=limit)
+    return templates.TemplateResponse("home.html", {
+        "request": request,
+        "data": data,
+        "sync_data_url": "/coronavirus/sync_coronavirus_data/jhu"
+    })

coronavirus/schemas.py

@@ -4,7 +4,6 @@
 
 from datetime import date as date_
 from datetime import datetime
-from typing import List
 
 from pydantic import BaseModel
 
@@ -21,7 +20,6 @@ class CreateCity(BaseModel):
     country: str
     country_code: str
     country_population: int
-    data: List[CreateData] = []
 
 
 class ReadData(CreateData):