-
Notifications
You must be signed in to change notification settings - Fork 1
/
admapi.py
108 lines (95 loc) · 3.88 KB
/
admapi.py
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
from fastapi import FastAPI, Header, Body, Request, HTTPException
from fastapi import status as httpcode
from fastapi.responses import JSONResponse, HTMLResponse
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
from modelAdm import PenAdmDownloadResponseModel
from modelAdm import PenAdmCopyReqModel
from modelAdm import PenAdmStatAckModel
import asyncio
import aiohttp
import util
class PenAdmStat(BaseModel):
epoch: datetime
#uptime: str
def api(config):
logger = config.logger
fastapi_kwargs = {}
#fastapi_kwargs = dict(docs_url=None, redoc_url=None, openapi_url=None)
app = FastAPI(**fastapi_kwargs)
stat = PenAdmStat(epoch=datetime.now())
# initializing house keeper
#hk = Housekeepter.CodeWordMap("codeNwords/word_list.txt")
# status reporter
async def lab_copy_worker():
while True:
# get items from DB. /4/list
# put items to Lab DB.
await asyncio.sleep(config.lab_copy_interval)
logger.info(f"COPY: nb_items=0")
config.loop.create_task(lab_copy_worker())
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(
request: Request, exc: RequestValidationError
) -> JSONResponse:
content = jsonable_encoder(exc.errors())
logger.error(f"Model Validation Error: {content}")
return JSONResponse(
status_code=httpcode.HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": content},
)
#
# API
#
@app.get(
"/dl/{c3w_words}",
response_description="Download the patient data specified by 3-word-code.",
response_model=PenAdmDownloadResponseModel,
status_code=httpcode.HTTP_200_OK,
)
async def get_patient_data(request: Request, c3w_words: str):
logger.debug(f"APP get_patient_data by c3w_words: {c3w_words}")
logger.debug(f"Accessed: {request.client.host}")
if (len(config.allow_ip_addrs) > 0 and
request.client.host not in config.allow_ip_addrs):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
detail=f"{request.client.host} not allowed")
logger.debug(f"get_patient_data: {c3w_words}")
url = f"{config.db_api_url}/a/c3ww/{c3w_words}"
status, ctype, content = await util.get_item(
url, logger, config.enable_tls)
if status != httpcode.HTTP_200_OK:
raise HTTPException(status_code=httpcode.HTTP_404_NOT_FOUND,
detail=f"c3w_words not found: {c3w_words}")
if ctype != "application/json":
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"invalid content-type (system)")
logger.debug(f"RET: data={content}")
# update tsStep3 and post DB
in_json = {
"pid": content["pid"],
"tsStep3": util.get_timestamp(config.tz)
}
logger.debug(f"POST DB: trying: {in_json}")
url = f"{config.db_api_url}/3"
status, content = await util.post_item(url, in_json,
enable_tls=config.enable_tls)
if status != httpcode.HTTP_200_OK:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to update tsStep3 for {c3w_words}")
return content
@app.get(
"/stat",
response_description="Get the stat.",
response_model=PenAdmStatAckModel,
status_code=httpcode.HTTP_200_OK
)
async def get_stat(request: Request):
logger.debug(f"APP get_stat")
return { "uptime": "__test__" }
return app