-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
185 lines (152 loc) · 5.98 KB
/
app.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
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
from datetime import timedelta
from typing import Optional
from fastapi import Depends, FastAPI, Form, HTTPException, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from fastapi.security import OAuth2PasswordRequestForm
from passlib.context import CryptContext
import uvicorn
import db
from db import Device
from pipeline import FunctionNotFoundError, FunctionParamUnmatchError, Pipeline
import user_authorization as user_auth
hashed_api_password = b'$2b$12$3jMfq3IMzFOzJ.LqXiaelOBKbU4A7n.LyBKNAR39lTyKF44WcPscK'
pw_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get('/')
async def root() -> FileResponse:
return FileResponse('static/index.html')
@app.get('/json/signals')
def json_last_signal_ts() -> JSONResponse:
devices: list[Device] = db.select_devices()
devices = [device.dict() for device in devices]
for device in devices:
device['last_heartbeat_timestamp'] = str(device['last_heartbeat_timestamp'])
retval = {
'devices': devices,
'heartbeat_log': db.select_heartbeat_log_summation(),
}
return JSONResponse(jsonable_encoder(retval))
@app.post('/devices/active')
def update_is_active(
user: user_auth.UserInDB = Depends(user_auth.get_current_user),
device_name: str = Form(...),
is_active: str = Form(...),
) -> PlainTextResponse:
if not user: # User authorization
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
active = False if is_active.lower() == 'false' else True
db.update_is_active(device_name, active)
return PlainTextResponse('successfully registered\n', status_code=200)
@app.post('/api/heartbeat')
def api_heartbeat(
password: str = Form(...),
name: str = Form(...),
nvidia_smi: Optional[str] = Form(None)
) -> PlainTextResponse:
if not pw_context.verify(password, hashed_api_password): # check credential
return PlainTextResponse(content='invalid password\n', status_code=403)
try:
db.post_heartbeat(name, nvidia_smi)
db.insert_heartbeat_log(name)
except ValueError:
return PlainTextResponse(content='invalid name\n', status_code=400)
return_message = db.select_return_message(name)
try:
content = Pipeline.feed(return_message)
except (FunctionNotFoundError, FunctionParamUnmatchError):
content = return_message
return PlainTextResponse(
content=content,
status_code=200
)
@app.post('/api/v2/heartbeat')
def api_heartbeat(
password: str = Form(...),
device_name: str = Form(...),
report: Optional[str] = Form(None)
) -> PlainTextResponse:
if not pw_context.verify(password, hashed_api_password): # check credential
return PlainTextResponse(content='invalid password\n', status_code=403)
try:
db.post_heartbeat(device_name, report)
db.insert_heartbeat_log(device_name)
except ValueError:
return PlainTextResponse(content='invalid name\n', status_code=400)
return_message = db.select_return_message(device_name)
try:
content = Pipeline.feed(return_message)
except (FunctionNotFoundError, FunctionParamUnmatchError):
content = return_message
return PlainTextResponse(
content=content,
status_code=200
)
@app.post('/api/return_message')
def api_register_return_message(
user: user_auth.UserInDB = Depends(user_auth.get_current_user),
name: str = Form(...),
return_message: str = Form(...)
) -> PlainTextResponse|HTTPException:
if not user: # User authorization
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
# Process Special str
if return_message == '#empty':
return_message = ''
db.update_return_message(name, return_message)
return PlainTextResponse('successfully registered\n', status_code=200)
@app.post('/api/v2/register/device')
def api_register_device(
user: user_auth.UserInDB = Depends(user_auth.get_current_user),
device_name: str = Form(...),
) -> PlainTextResponse|HTTPException:
if not user: # User authorization
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
if len(device_name) < 3:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Device name must be more than 3 characters.",
)
try:
db.register_device(device_name, None, None)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="This device is already registered.",
)
return PlainTextResponse('successfully registered\n', status_code=200)
@app.post('/api/token')
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends()
) -> JSONResponse|HTTPException:
user: user_auth.UserInDB|bool = user_auth.authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=user_auth.ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = user_auth.create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return JSONResponse ({
"access_token": access_token,
"token_type": "bearer",
})
if __name__ == '__main__':
uvicorn.run('app:app', host='127.0.0.1', port=8000)