Coverage for toardb / data / data.py: 65%

1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH 

2# SPDX-License-Identifier: MIT 

3 

4""" 

5Simple test API for data management 

6""" 

7 

8import json 

9from typing import List 

10from fastapi import APIRouter, Depends, HTTPException, \ 

11 File, UploadFile, Request 

12from sqlalchemy.orm import Session 

13from sqlalchemy.engine import Engine 

14from . import crud, schemas 

15from toardb.utils.database import ToarDbSession, engine, get_engine, get_db 

16from toardb.utils.utils import ( 

17 get_data_download_access_rights, 

18 get_map_data_download_access_rights, 

19 get_admin_access_rights, 

20 get_data_change_access_rights 

21) 

22from toardb.utils.settings import request_limitations as limits 

23import datetime as dt 

24 

25router = APIRouter() 

26 

27# CRUD: create, retrieve, update, delete 

28 

29# 1. retrieve 

30 

31#get all data of one timeseries 

32@router.get('/data/timeseries/{timeseries_id}', response_model=schemas.Composite, response_model_exclude_unset=True, response_model_exclude_none=True) 

33def get_data(timeseries_id: int, request: Request, 

34 access: dict = Depends(get_data_download_access_rights), 

35 db: Session = Depends(get_db)): 

36 # check whether the user is sending the request via the dashboard 

37 if not access['lfromdashboard']: 

38 if access['status_code'] == 200: 

39 if access['role'] == 'anonymous' and not limits['anonymous']['min_tsid'] <= timeseries_id <= limits['anonymous']['max_tsid']: 

40 raise HTTPException(status_code=401, detail=f"Anonymous access only to timeseries within the ID range of {limits['anonymous']['min_tsid']} to {limits['anonymous']['max_tsid']}.") 

41 else: 

42 raise HTTPException(status_code=401, detail="Unauthorized.") 

43 

44 db_data = crud.get_data(db, timeseries_id=timeseries_id, path_params=request.path_params, query_params=request.query_params) 

45 if db_data is None: 

46 raise HTTPException(status_code=404, detail="Data not found.") 

47 return db_data 

48 

49 

50#get all data of one timeseries 

51@router.get('/data/timeseries/id/{timeseries_id}', response_model=schemas.Composite, response_model_exclude_unset=True, response_model_exclude_none=True) 

52def get_data2(timeseries_id: int, request: Request, 

53 access: dict = Depends(get_data_download_access_rights), 

54 db: Session = Depends(get_db)): 

55 if access['status_code'] == 200: 

56 if access['role'] == 'anonymous' and not limits['anonymous']['min_tsid'] <= timeseries_id <= limits['anonymous']['max_tsid']: 

57 raise HTTPException(status_code=401, detail=f"Anonymous access only to timeseries within the ID range of {limits['anonymous']['min_tsid']} to {limits['anonymous']['max_tsid']}.") 

58 

59 db_data = crud.get_data(db, timeseries_id=timeseries_id, path_params=request.path_params, query_params=request.query_params) 

60 if db_data is None: 

61 raise HTTPException(status_code=404, detail="Data not found.") 

62 return db_data 

63 else: 

64 raise HTTPException(status_code=401, detail="Unauthorized.") 

65 

66 

67#get all data of one timeseries (including staging data) 

68@router.get('/data/timeseries_with_staging/id/{timeseries_id}', response_model=schemas.Composite, response_model_exclude_unset=True, response_model_exclude_none=True) 

69def get_data_with_staging(timeseries_id: int, flags: str = None, format: str = 'json', 

70 access: dict = Depends(get_data_download_access_rights), 

71 db: Session = Depends(get_db)): 

72 if access['status_code'] == 200: 

73 if access['role'] == 'anonymous' and not limits['anonymous']['min_tsid'] <= timeseries_id <= limits['anonymous']['max_tsid']: 

74 raise HTTPException(status_code=401, detail=f"Anonymous access only to timeseries within the ID range of {limits['anonymous']['min_tsid']} to {limits['anonymous']['max_tsid']}.") 

75 

76 db_data = crud.get_data_with_staging(db, timeseries_id=timeseries_id, flags=flags, format=format) 

77 if db_data is None: 

78 raise HTTPException(status_code=404, detail="Data not found.") 

79 return db_data 

80 else: 

81 raise HTTPException(status_code=401, detail="Unauthorized.") 

82 

83 

84#get map data (for a special variable and timestamp) 

85@router.get('/data/map/') 

86def get_map_data(variable_id: int = 5, daterange: str = '2023-02-22 12:00,2023-02-22 12:00', 

87 access: dict = Depends(get_map_data_download_access_rights), 

88 db: Session = Depends(get_db)): 

89 # check whether the user is sending the request via the dashboard 

90 if not access['lfromdashboard']: 

91 if access['status_code'] != 200: 

92 raise HTTPException(status_code=401, detail="Unauthorized.") 

93 db_data = crud.get_map_data(db, variable_id=variable_id, daterange=daterange) 

94 if db_data is None: 

95 raise HTTPException(status_code=404, detail="Data not found.") 

96 return db_data 

97 

98 

99#get the next available version for one timeseries 

100@router.get('/data/timeseries/next_version/{timeseries_id}') 

101def get_version(timeseries_id: int, request: Request, db: Session = Depends(get_db)): 

102 version = crud.get_next_version(db, timeseries_id=timeseries_id, path_params=request.path_params, query_params=request.query_params) 

103 return version 

104 

105 

106@router.get('/data/timeseries_merged/', response_model=schemas.Composite, response_model_exclude_unset=True, response_model_exclude_none=True) 

107def get_merged_data(variable_id: int, station_code: str, request: Request, daterange: str = None, has_role: str = None, 

108 access: dict = Depends(get_data_download_access_rights), 

109 db: Session = Depends(get_db)): 

110 # check whether the user is sending the request via the dashboard 

111 if not access['lfromdashboard']: 

112 if access['status_code'] == 200: 

113 if access['role'] == 'anonymous': 

114 merging_list = crud.get_merging_list(db, variable_id=variable_id, station_code=station_code, daterange=daterange, role=has_role) 

115 # gather all timeseries_ids involved in this request 

116 ts_ids = set() 

117 if merging_list[0] != []: 

118 [ts_ids.add(ts_id[2]) for ts_id in merging_list[0]] 

119 if merging_list[1] != []: 

120 [ts_ids.add(ts_id) for ts_id in merging_list[1]] 

121 min_request_id = -1 if not ts_ids else min(ts_ids) 

122 max_request_id = -1 if not ts_ids else max(ts_ids) 

123 if (not limits['anonymous']['min_tsid'] <= min_request_id or 

124 not max_request_id <= limits['anonymous']['max_tsid']): 

125 raise HTTPException(status_code=401, detail=f"Anonymous access only to timeseries within the ID range of {limits['anonymous']['min_tsid']} to {limits['anonymous']['max_tsid']}.") 

126 else: 

127 raise HTTPException(status_code=401, detail="Unauthorized.") 

128 

129 db_data = crud.get_merged_data(db, variable_id=variable_id, station_code=station_code, role=has_role, path_params=request.path_params, query_params=request.query_params) 

130 if db_data is None: 

131 raise HTTPException(status_code=404, detail="Data not found.") 

132 return db_data 

133 

134 

135@router.get('/data/get_merging_list/') 

136def get_merging_list(variable_id: int, station_code: str, daterange: str, has_role: str = None, db: Session = Depends(get_db)): 

137 return crud.get_merging_list(db, variable_id=variable_id, station_code=station_code, daterange=daterange, role=has_role) 

138 

139 

140#get map data (for a special variable and timestamp) 

141@router.get('/data/map/') 

142def get_map_data(variable_id: int = 5, daterange: str = '2023-02-22 12:00,2023-02-22 12:00', db: Session = Depends(get_db)): 

143 db_data = crud.get_map_data(db, variable_id=variable_id, daterange=daterange) 

144 if db_data is None: 

145 raise HTTPException(status_code=404, detail="Data not found.") 

146 return db_data 

147 

148 

149# post and patch only via authorization by Helmholtz AAI 

150 

151# 2. create 

152 

153@router.post('/data/timeseries/') 

154def create_data(file: UploadFile = File(...), 

155 toarqc_config_type: str = 'standard', 

156 dry_run: bool = False, 

157 force: bool = False, 

158 access: dict = Depends(get_admin_access_rights), 

159 db: Session = Depends(get_db), 

160 engine: Engine = Depends(get_engine)): 

161# # the next three lines are automatically done by database management, 

162# # but we do want helpful error messages! 

163# db_data = crud.get_data_by_datetime_and_timeseriesid(db, datetime=data.datetime, timeseries_id=data.timeseries_id) 

164# if db_data: 

165# raise HTTPException(status_code=400, detail="Data already registered.") 

166 

167# BUT: 

168# we want to upload a whole file! 

169# 

170 # check whether the post is authorized (401: Unauthorized) 

171 if access['status_code'] == 200: 

172 response = crud.create_data(db, engine, author_id=access['auth_user_id'], input_handle=file, 

173 toarqc_config_type=toarqc_config_type, dry_run=dry_run, force=force) 

174 if response.status_code != 200 and response.status_code != 446: 

175 msg = json.loads(response.body.decode('utf-8')) 

176 # try to parse error messages from DBS (to be more understandable) 

177 msg2 = "An error occurred in public.data insertion: <class 'psycopg2.errors.UniqueViolation'>" 

178 if (msg == msg2): 

179 msg = 'Data for timeseries already registered.' 

180 raise HTTPException(status_code=400, detail=msg) 

181 return response 

182 else: 

183 raise HTTPException(status_code=401, detail="Unauthorized.") 

184 

185 

186@router.post('/data/timeseries/bulk/') 

187def create_bulk_data(bulk: List[schemas.DataCreate], 

188 toarqc_config_type: str = 'standard', 

189 dry_run: bool = False, 

190 force: bool = False, 

191 access: dict = Depends(get_admin_access_rights), 

192 db: Session = Depends(get_db), 

193 engine: Engine = Depends(get_engine)): 

194 # check whether the post is authorized (401: Unauthorized) 

195 if access['status_code'] == 200: 

196 response = crud.create_bulk_data(db, engine, bulk=bulk, 

197 author_id=access['auth_user_id'], toarqc_config_type=toarqc_config_type, dry_run=dry_run, 

198 force=force) 

199 if response.status_code != 200: 

200 msg = response.body.decode('utf-8') 

201 # try to parse error messages from DBS (to be more understandable) 

202 msg2 = '"An error occurred in public.data insertion: <class \'psycopg2.errors.UniqueViolation\'>"' 

203 if (msg == msg2): 

204 msg = 'Data for timeseries already registered.' 

205 raise HTTPException(status_code=400, detail=msg) 

206 return response 

207 else: 

208 raise HTTPException(status_code=401, detail="Unauthorized.") 

209 

210 

211@router.post('/data/timeseries/record/') 

212def create_data_record(series_id: int, 

213 datetime: dt.datetime, 

214 value: float, 

215 flag: str, 

216 version: str = None, 

217 suppress_unique_violation: bool = False, 

218 access: dict = Depends(get_admin_access_rights), 

219 db: Session = Depends(get_db), 

220 engine: Engine = Depends(get_engine)): 

221 # check whether the post is authorized (401: Unauthorized) 

222 if access['status_code'] == 200: 

223 response = crud.create_data_record(db, engine, series_id=series_id, datetime=datetime, 

224 value=value, flag=flag, version=version, 

225 author_id=access['auth_user_id']) 

226 if response.status_code != 200: 

227 msg = response.body.decode('utf-8') 

228 # try to parse error messages from DBS (to be more understandable) 

229 msg2 = '"An error occurred in data insertion: <class \'psycopg2.errors.UniqueViolation\'>"' 

230 if (msg == msg2): 

231 if not suppress_unique_violation: 

232 msg = 'Data for timeseries already registered.' 

233 raise HTTPException(status_code=400, detail=msg) 

234 else: 

235 raise HTTPException(status_code=400, detail=msg) 

236 return response 

237 else: 

238 raise HTTPException(status_code=401, detail="Unauthorized.") 

239 

240# 3. update 

241 

242@router.patch('/data/timeseries/') 

243def patch_data(description: str, 

244 version: str, 

245 file: UploadFile = File(...), 

246 toarqc_config_type: str = 'standard', 

247 force: bool = False, 

248 access: dict = Depends(get_data_change_access_rights), 

249 db: Session = Depends(get_db), 

250 engine: Engine = Depends(get_engine)): 

251 # check whether the patch is authorized (401: Unauthorized) 

252 if access['status_code'] == 200: 

253 return crud.patch_data(db, engine, description=description, version=version, toarqc_config_type=toarqc_config_type, 

254 force=force, author_id=access['auth_user_id'], input_handle=file) 

255 else: 

256 raise HTTPException(status_code=401, detail="Unauthorized.") 

257 

258@router.patch('/data/timeseries/bulk/') 

259def patch_bulk_data(description: str, 

260 version: str, 

261 bulk: List[schemas.DataPatch], 

262 toarqc_config_type: str = 'standard', 

263 force: bool = False, 

264 no_archive: bool = False, 

265 access: dict = Depends(get_data_change_access_rights), 

266 db: Session = Depends(get_db), 

267 engine: Engine = Depends(get_engine)): 

268 # check whether the patch is authorized (401: Unauthorized) 

269 if access['status_code'] == 200: 

270 return crud.patch_bulk_data(db, engine, description=description, version=version, bulk=bulk, author_id=access['auth_user_id'], 

271 toarqc_config_type=toarqc_config_type, force=force, no_archive=no_archive) 

272 else: 

273 raise HTTPException(status_code=401, detail="Unauthorized.") 

274