Coverage for toardb / timeseries / crud.py: 77%

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

2# SPDX-License-Identifier: MIT 

3 

4""" 

5Create, Read, Update, Delete functionality 

6 

7""" 

8from sqlalchemy import insert, select, and_, func, text 

9from sqlalchemy.orm import Session, load_only 

10from sqlalchemy.util import _collections 

11from sqlalchemy.exc import IntegrityError 

12from geoalchemy2.elements import WKBElement, WKTElement 

13from fastapi import File, UploadFile 

14from fastapi.responses import JSONResponse 

15import datetime as dt 

16import json 

17from . import models 

18from .models import TimeseriesChangelog, timeseries_timeseries_roles_table, \ 

19 timeseries_timeseries_annotations_table, s1_contributors_table, YearlyCoverage 

20from toardb.stationmeta.models import StationmetaCore, StationmetaGlobal 

21from toardb.stationmeta.schemas import get_coordinates_from_geom, get_geom_from_coordinates, get_coordinates_from_string 

22from toardb.stationmeta.crud import get_stationmeta_by_id, get_stationmeta_core, station_id_exists, get_stationmeta_changelog 

23from toardb.contacts.crud import get_organisation_by_name, get_contact 

24from toardb.contacts.models import Organisation, Person, Contact 

25from toardb.contacts.schemas import ContactBase 

26from toardb.variables.models import Variable 

27from toardb.variables.crud import get_variable 

28from .schemas import TimeseriesCreate, TimeseriesPatch, TimeseriesRoleNoCreate, TimeseriesRoleFields 

29from toardb.utils.utils import get_value_from_str, get_str_from_value, create_filter, roles_params 

30import toardb 

31 

32 

33def clean_additional_metadata(ad_met_dict): 

34 # all changes are permanent! 

35 if not isinstance(ad_met_dict,dict): 

36 tmp = ad_met_dict.replace('"','\\"') 

37 return tmp.replace("'",'"') 

38 # there is a mismatch with additional_metadata 

39 additional_metadata = ad_met_dict 

40 for key, value in additional_metadata.items(): 

41 if isinstance(value,dict): 

42 for key2, value2 in value.items(): 

43 if isinstance(value2,str): 

44 additional_metadata[key][key2] = value2.replace("'","$apostroph$") 

45 elif isinstance(value,str): 

46 additional_metadata[key] = value.replace("'","$apostroph$") 

47 additional_metadata = str(additional_metadata).replace('"','\\"') 

48 additional_metadata = str(additional_metadata).replace("'",'"') 

49 additional_metadata = str(additional_metadata).replace("$apostroph$","'") 

50 return additional_metadata 

51 

52 

53def get_timeseries(db: Session, timeseries_id: int, fields: str = None): 

54 if fields: 

55 fields = ','.join('"{}"'.format(word) for word in fields.split(',')) 

56 # this command returns a tuple (which is unmutable --> problems with additional_metadata and coordinates!) 

57 # also this command could not deal with reserved words used as column names (like "order") 

58 resultproxy = db.execute(f'SELECT {fields} FROM timeseries WHERE id={timeseries_id} LIMIT 1') 

59 db_object_dict = [ rowproxy._asdict() for rowproxy in resultproxy ][0] 

60 if fields.find('additional_metadata') != -1: 

61 db_object_dict['additional_metadata'] = clean_additional_metadata(db_object_dict['additional_metadata']) 

62 db_object = models.Timeseries(**db_object_dict) 

63 else: 

64 db_object = db.query(models.Timeseries).filter(models.Timeseries.id == timeseries_id).first() 

65 if db_object: 

66 # only for internal use! 

67 if db_object.data_license_accepted: 

68 del db_object.data_license_accepted 

69 if db_object.dataset_approved_by_provider: 

70 del db_object.dataset_approved_by_provider 

71 if db_object: 

72 try: 

73 # there is a mismatch with additional_metadata 

74 db_object.additional_metadata = clean_additional_metadata(db_object.additional_metadata) 

75 except: 

76 pass 

77 try: 

78 # there is also a mismatch with coordinates and additional_metadata from station object 

79 if isinstance(db_object.station.coordinates, (WKBElement, WKTElement)): 

80 db_object.station.coordinates = get_coordinates_from_geom(db_object.station.coordinates) 

81 # there is a mismatch with additional_metadata 

82 if isinstance(db_object.station.additional_metadata, dict): 

83 db_object.station.additional_metadata = json.dumps(db_object.station.additional_metadata) 

84 except: 

85 pass 

86 return db_object 

87 

88# https://www.reddit.com/r/flask/comments/ypqk40/default_datetimenow_not_using_current_time/ 

89# dt.datetime.now(dt.timezone.utc) cannot be used as default value, since it will already be 

90# evaluated at the start of the worker process and **not** when actually accessing the data 

91# def get_citation(db: Session, timeseries_id: int, datetime: dt.datetime = dt.datetime.now(dt.timezone.utc)): 

92def get_citation(db: Session, timeseries_id: int, datetime: dt.datetime = None): 

93 if not datetime: 

94 datetime = dt.datetime.now(dt.timezone.utc) 

95 db_object = db.query(models.Timeseries).filter(models.Timeseries.id == timeseries_id).first() 

96 # there is a mismatch with additional_metadata 

97 PI = "unknown" 

98 originators = False 

99 attribution = None 

100 if db_object: 

101 pi_role = get_value_from_str(toardb.toardb.RC_vocabulary,'PrincipalInvestigator') 

102 originator_role = get_value_from_str(toardb.toardb.RC_vocabulary,'Originator') 

103 contributor_role = get_value_from_str(toardb.toardb.RC_vocabulary,'Contributor') 

104 list_of_originators = [] 

105 for db_role in db_object.roles: 

106 if (db_role.role == pi_role): 

107 db_contact = get_contact(db, contact_id = db_role.contact_id) 

108 PI = db_contact.name 

109 elif (db_role.role == originator_role): 

110 originators = True 

111 db_contact = get_contact(db, contact_id = db_role.contact_id) 

112 list_of_originators.append(db_contact.name) 

113 elif (db_role.role == contributor_role): 

114 db_contact = get_contact(db, contact_id = db_role.contact_id) 

115 list_of_originators.append(db_contact.name) 

116 list_of_data_originators = ", ".join(list_of_originators) 

117 # if no PI is given, the resource provider's longname should be named 

118 if PI == "unknown": 

119 role = get_value_from_str(toardb.toardb.RC_vocabulary,'ResourceProvider') 

120 for db_role in db_object.roles: 

121 if (db_role.role == role): 

122 db_contact = get_contact(db, contact_id = db_role.contact_id) 

123 PI = db_contact.longname 

124 if db_contact.attribution != '': 

125 attribution = db_contact.attribution 

126 var = get_variable(db, variable_id=db_object.variable_id).name 

127 station = get_stationmeta_by_id(db, station_id=db_object.station_id).name 

128 dataset_version = db_object.provider_version.strip() 

129 dataset_doi = db_object.doi.strip() 

130 citation = f"{PI}: time series of {var} at {station}, accessed from the TOAR database on {datetime}" 

131 if attribution and list_of_data_originators: 

132 attribution = attribution.format(orga_or_origs="data originators" if originators else "contributing organisations", list_of_data_originators=list_of_data_originators) 

133 if dataset_version != 'N/A': 

134 citation += f", original dataset version {dataset_version}" 

135 if dataset_doi!= '': 

136 citation += f", original dataset doi: {dataset_doi}" 

137 license_txt = "This data is published under a Creative Commons Attribution 4.0 International (CC BY 4.0). https://creativecommons.org/licenses/by/4.0/" 

138 return {"attribution": attribution, "citation": citation, "license": license_txt} 

139 

140def adapt_db_object(db_object, db, fields=False, lconstr_roles=False): 

141 if fields: 

142 db_object = dict(zip((field for field in fields if field not in {"station_changelog", "changelog"}), db_object)) 

143 

144 # there is a mismatch with coordinates and additional_metadata 

145 if "coordinates" in db_object: 

146 db_object["coordinates"] = get_coordinates_from_string(db_object["coordinates"]) 

147 

148 if "additional_metadata" in db_object: 

149 db_object["additional_metadata"] = clean_additional_metadata(db_object["additional_metadata"]) 

150 

151 if "station_id" in db_object: 

152 station_id = {"id": db_object["station_id"]} 

153 db_object["station"] = station_id 

154 del db_object["station_id"] 

155 

156 if "station_additional_metadata" in db_object: 

157 cleaned_metadata = clean_additional_metadata(db_object["station_additional_metadata"]) 

158 if "station" not in db_object: 

159 db_object["station"] = {} 

160 db_object["station"]["additional_metadata"] = cleaned_metadata 

161 del db_object["station_additional_metadata"] 

162 

163 if "variable_id" in db_object: 

164 variable_id = {"id": db_object["variable_id"]} 

165 db_object["variable"] = variable_id 

166 del db_object["variable_id"] 

167 

168 if "changelog" in db_object: 

169 db_object["changelog"] = get_timeseries_changelog(db, db_object["id"]) 

170 

171 if "station_changelog" in db_object: 

172 try: 

173 db_object["station_changelog"] = get_stationmeta_changelog(db, db_object["station_id"]) 

174 except Exception: 

175 pass 

176 

177 if lconstr_roles: 

178 # example, how to put the roles explicitly (not needed at the moment) 

179 # organisation = get_contact(db, contact_id=39) 

180 # roles_atts["contact"] = {"id": 39, "organisation": organisation.__dict__} 

181 roles_atts = {key: value for key, value in db_object.items() if key in roles_params} 

182 db_object = {key: value for key, value in db_object.items() if key not in roles_params} 

183 db_object["roles"] = TimeseriesRoleFields(**roles_atts) 

184 else: 

185 if isinstance(db_object.station.coordinates, (WKBElement, WKTElement)): 

186 db_object.station.coordinates = get_coordinates_from_geom(db_object.station.coordinates) 

187 # there is a mismatch with additional_metadata 

188 if isinstance(db_object.station.additional_metadata, dict): 

189 db_object.station.additional_metadata = json.dumps(db_object.station.additional_metadata) 

190 db_object.additional_metadata = clean_additional_metadata(db_object.additional_metadata) 

191 

192 

193 #Internall use 

194 try: 

195 del db_object.data_license_accepted 

196 except AttributeError: 

197 pass 

198 

199 try: 

200 del db_object.dataset_approved_by_provider 

201 except AttributeError: 

202 pass 

203 

204 return db_object 

205 

206class TimeseriesQuery: 

207 def __init__(self, sign, query, fields, lconstr_roles): 

208 self.sign = sign 

209 self.query = query 

210 self.fields = fields 

211 self.lconstr_roles = lconstr_roles 

212 

213 @staticmethod 

214 def aggregate(querys): 

215 aggregated_query = next(querys) 

216 for query in querys: 

217 if aggregated_query.fields != query.fields: 

218 raise ValueError("Fields of subquerys are diffrent") 

219 aggregated_query = TimeseriesQuery( 

220 True, 

221 aggregated_query.query.union(query.query) 

222 if query.sign 

223 else aggregated_query.query.except_(query.query), 

224 aggregated_query.fields, 

225 aggregated_query.lconstr_roles or query.lconstr_roles, 

226 ) 

227 return aggregated_query 

228 

229 @staticmethod 

230 def from_query_params(query_params, db, endpoint="timeseries", sign=True): 

231 limit, offset, fields, format, filters = create_filter(query_params, endpoint) 

232 t_filter = filters["t_filter"] 

233 t_r_filter = filters["t_r_filter"] 

234 s_c_filter = filters["s_c_filter"] 

235 s_g_filter = filters["s_g_filter"] 

236 

237 if fields: 

238 # If only certain fields are selected the return type is not a orm object anymore but a dict 

239 # sort input fields to be sure to replace station_changelog before changelog 

240 fields = sorted(fields.split(","), reverse=True) 

241 if "role" in fields: 

242 fields.remove("role") 

243 fields.extend(roles_params) 

244 

245 field_map = { 

246 "id": models.Timeseries.id, 

247 "order": models.Timeseries.order, 

248 "additional_metadata": models.Timeseries.additional_metadata, 

249 "station_id": StationmetaCore.id, 

250 "variable_id": Variable.id, 

251 "name": StationmetaCore.name, 

252 "coordinates": func.ST_AsText(StationmetaCore.coordinates), 

253 "station_country": StationmetaCore.country, 

254 "station_additional_metadata": StationmetaCore.additional_metadata 

255 } 

256 

257 query_select = [field_map.get(field, text(field)) for field in fields] 

258 

259 else: 

260 query_select = [models.Timeseries] 

261 

262 if 'station_id' in fields and not id in fields: 

263 query = ( 

264 db.query(*query_select) 

265 .select_from(models.Timeseries) 

266 .distinct() 

267 .filter(text(t_filter), text(s_c_filter), text(s_g_filter), text(t_r_filter)) 

268 .join(StationmetaCore) 

269 .join(StationmetaGlobal) 

270 .join(timeseries_timeseries_roles_table) 

271 .join(models.TimeseriesRole) 

272 .join(Contact) 

273 .join(Organisation) 

274 .join(Person) 

275 .join(Variable) 

276 .execution_options(stream_results=True) 

277 ) 

278 else: 

279 query = ( 

280 db.query(*query_select) 

281 .select_from(models.Timeseries) 

282 .distinct() 

283 .filter(text(t_filter), text(s_c_filter), text(s_g_filter), text(t_r_filter)) 

284 .join(StationmetaCore) 

285 .join(StationmetaGlobal) 

286 .join(timeseries_timeseries_roles_table) 

287 .join(models.TimeseriesRole) 

288 .join(Contact) 

289 .join(Organisation) 

290 .join(Person) 

291 .join(Variable) 

292 .execution_options(stream_results=True) 

293 .order_by(models.Timeseries.id) 

294 ) 

295 

296 # Apply NOT filter with role logic 

297 if "NOT" in t_r_filter: 

298 role_ids = get_role_id_from_string(db, query_params.get("has_role")[1:]) 

299 query = query.filter( 

300 ~models.Timeseries.id.in_( 

301 select(timeseries_timeseries_roles_table.c.timeseries_id).where( 

302 timeseries_timeseries_roles_table.c.role_id.in_(role_ids) 

303 ) 

304 ) 

305 ) 

306 

307 if limit: 

308 query = query.limit(limit).offset(offset) 

309 

310 return TimeseriesQuery(sign, query, fields, lconstr_roles=any(field in roles_params for field in fields)) 

311 

312 def adapt_objects(self, db): 

313 return [adapt_db_object(db_object, db, self.fields, self.lconstr_roles) for db_object in self.query] 

314 

315 

316def search_all(db, path_params, query_params, lts=False, endpoint=None): 

317 if endpoint == None: 

318 endpoint = "timeseries" if lts else "search" 

319 

320 try: 

321 ts_list = TimeseriesQuery.from_query_params(query_params, db, endpoint).adapt_objects(db) 

322 # remove duplicates 

323 if ts_list and isinstance(ts_list[0], dict): 

324 try: 

325 ts_set = set(json.dumps(ts, sort_keys=True) for ts in ts_list) 

326 return [json.loads(ts) for ts in ts_set] 

327 except: # not correct, because duplicates are NOT removed!!! 

328 return ts_list 

329 else: 

330 return ts_list 

331 except (KeyError, ValueError) as e: 

332 status_code = 400 

333 return JSONResponse(status_code=status_code, content=str(e)) 

334 

335 

336def search_all_aggregation(db, path_params, signs, query_params_list, lts=False): 

337 endpoint = "timeseries" if lts else "search" 

338 

339 try: 

340 ts_list = TimeseriesQuery.aggregate( 

341 TimeseriesQuery.from_query_params(query_params, db, endpoint, sign) 

342 for sign, query_params in zip(signs, query_params_list) 

343 ).adapt_objects(db) 

344 # remove duplicates 

345 if ts_list and isinstance(ts_list[0], dict): 

346 try: 

347 ts_set = set(json.dumps(ts, sort_keys=True) for ts in ts_list) 

348 return [json.loads(ts) for ts in ts_set] 

349 except: # not correct, because duplicates are NOT removed!!! 

350 return ts_list 

351 else: 

352 return ts_list 

353 except (KeyError, ValueError) as e: 

354 status_code = 400 

355 return JSONResponse(status_code=status_code, content=str(e)) 

356 

357 

358 

359def get_timeseries_by_unique_constraints(db: Session, station_id: int, variable_id: int, resource_provider: str = None, 

360 sampling_frequency: str = None, provider_version: str = None, data_origin_type: str = None, 

361 data_origin: str = None, sampling_height: float = None, label: str = None): 

362 """ 

363 Criteria taken from TOAR_TG_Vol02_Data_Processing, 'Step 14: Identify Time Series' 

364 Criterion 14.1: id of the corresponding station 

365 Criterion 14.2: variable id 

366 Criterion 14.3: role: resource_provider (organisation) 

367 Criterion 14.4: sampling_frequency 

368 Criterion 14.5: version number 

369 Criterion 14.6: data_origin_type (measurement or model) 

370 Criterion 14.7: data origin 

371 Criterion 14.8: sampling height 

372 Criterion 14.9: data filtering procedures or other special dataset identifiers (use database field 'label') 

373 """ 

374 

375# print("in get_timeseries_by_unique_constraints") 

376# print(f"station_id: {station_id}, variable_id: {variable_id}, resource_provider: {resource_provider}, ", \ 

377# f"sampling_frequency: {sampling_frequency}, provider_version: {provider_version}, data_origin_type: {data_origin_type}, ", \ 

378# f"data_origin: {sampling_frequency}, sampling_height: {sampling_height}, label: {label}") 

379 

380 # filter for criterion 14.1 and 14.2 

381 ret_db_object = db.query(models.Timeseries).filter(models.Timeseries.station_id == station_id) \ 

382 .filter(models.Timeseries.variable_id == variable_id).all() 

383 # if already not found: return None 

384 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

385 if len(ret_db_object) == 0: 

386 return None 

387 

388 

389 # filter for criterion 14.3 

390 if resource_provider: 

391 # issue with '/' in organisation longname ==> only possible with double encoding! 

392 resource_provider = resource_provider.replace('%2F', '/') 

393 role_num = get_value_from_str(toardb.toardb.RC_vocabulary,'ResourceProvider') 

394 iter_obj = ret_db_object.copy() 

395 counter=0 

396 for db_object in iter_obj: 

397 found = False 

398 for role in db_object.roles: 

399 # resource provider is always an organisation! 

400 organisation = get_contact(db, contact_id=role.contact_id) 

401 if ((role_num == role.role) and (organisation.longname == resource_provider)): 

402 found = True 

403 if not found: 

404 ret_db_object.pop(counter) 

405 else: 

406 counter += 1 

407 else: 

408 # time series that do not have a resource_provider are not identical to those who do not! 

409 role_num = get_value_from_str(toardb.toardb.RC_vocabulary,'ResourceProvider') 

410 iter_obj = ret_db_object.copy() 

411 counter=0 

412 for db_object in iter_obj: 

413 found = False 

414 for role in db_object.roles: 

415 if (role_num == role.role): 

416 found = True 

417 if found: 

418 ret_db_object.pop(counter) 

419 else: 

420 counter += 1 

421 

422 

423 # if already only none object --> return 

424 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

425 if len(ret_db_object) == 0: 

426 return None 

427 

428 # filter for criterion 14.4 

429 if sampling_frequency: 

430 iter_obj = ret_db_object.copy() 

431 counter=0 

432 for db_object in iter_obj: 

433 if not (db_object.sampling_frequency == sampling_frequency): 

434 ret_db_object.pop(counter) 

435 else: 

436 counter += 1 

437 

438 # if already only none object --> return 

439 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

440 if len(ret_db_object) == 0: 

441 return None 

442 

443 # filter for criterion 14.5 

444 if provider_version: 

445 iter_obj = ret_db_object.copy() 

446 counter=0 

447 for db_object in iter_obj: 

448 if not (db_object.provider_version == provider_version): 

449 ret_db_object.pop(counter) 

450 else: 

451 counter += 1 

452 

453 # if already only none object --> return 

454 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

455 if len(ret_db_object) == 0: 

456 return None 

457 

458 # filter for criterion 14.6 

459 if data_origin_type: 

460 data_origin_type_num = get_value_from_str(toardb.toardb.OT_vocabulary,data_origin_type) 

461 iter_obj = ret_db_object.copy() 

462 counter=0 

463 for db_object in iter_obj: 

464 if not (db_object.data_origin_type == data_origin_type_num): 

465 ret_db_object.pop(counter) 

466 else: 

467 counter += 1 

468 

469 # if already only none object --> return 

470 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

471 if len(ret_db_object) == 0: 

472 return None 

473 

474 # filter for criterion 14.7 

475 if data_origin: 

476 data_origin_num = get_value_from_str(toardb.toardb.DO_vocabulary,data_origin) 

477 iter_obj = ret_db_object.copy() 

478 counter=0 

479 for db_object in iter_obj: 

480 if not (db_object.data_origin == data_origin_num): 

481 ret_db_object.pop(counter) 

482 else: 

483 counter += 1 

484 

485 # if already only none object --> return 

486 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

487 if len(ret_db_object) == 0: 

488 return None 

489 

490 # filter for criterion 14.8 

491 if sampling_height: 

492 iter_obj = ret_db_object.copy() 

493 counter=0 

494 for db_object in iter_obj: 

495 if not (db_object.sampling_height == sampling_height): 

496 ret_db_object.pop(counter) 

497 else: 

498 counter += 1 

499 

500 # if already only none object --> return 

501 # if only one single object is found, it has to be checked whether all criterions are fullfilled 

502 if len(ret_db_object) == 0: 

503 return None 

504 

505 # filter for criterion 14.9 

506 if label: 

507 iter_obj = ret_db_object.copy() 

508 counter=0 

509 for db_object in iter_obj: 

510 if not (db_object.label == label): 

511 ret_db_object.pop(counter) 

512 else: 

513 counter += 1 

514 

515 # check that only one object is left!!! 

516 # adapt mismatches for return value 

517 if len(ret_db_object) == 0: 

518 ret_db_object = None 

519 else: 

520 if len(ret_db_object) == 1: 

521 ret_db_object = ret_db_object[0] 

522 # there is a mismatch with additional_metadata 

523 ret_db_object.additional_metadata = clean_additional_metadata(ret_db_object.additional_metadata) 

524 # there is also a mismatch with coordinates and additional_metadata from station object 

525 if isinstance(ret_db_object.station.coordinates, (WKBElement, WKTElement)): 

526 ret_db_object.station.coordinates = get_coordinates_from_geom(ret_db_object.station.coordinates) 

527 # there is a mismatch with additional_metadata 

528 if isinstance(ret_db_object.station.additional_metadata, dict): 

529 ret_db_object.station.additional_metadata = json.dumps(ret_db_object.station.additional_metadata) 

530 else: 

531 status_code=405 

532 message=f"Timeseries not unique, more criteria need to be defined." 

533 return JSONResponse(status_code=status_code, content=message) 

534 

535 return ret_db_object 

536 

537 

538def get_timeseries_changelog(db: Session, timeseries_id: int): 

539 return db.query(models.TimeseriesChangelog).filter(models.TimeseriesChangelog.timeseries_id == timeseries_id).all() 

540 

541 

542def get_timeseries_programme(db: Session, name: str): 

543 return db.query(models.TimeseriesProgramme).filter(models.TimeseriesProgramme.name == name).all() 

544 

545 

546# is this internal, or should this also go to public REST api? 

547# do we need this at all? 

548def get_role_ids_of_timeseries(db: Session, timeseries_id: int): 

549 db_objects = db.query(models.TimeseriesTimeseriesRoles) \ 

550 .filter(models.TimeseriesTimeseriesRoles.timeseries_id == timeseries_id) \ 

551 .all() 

552 return db_objects 

553 

554 

555# is this internal, or should this also go to public REST api? 

556def get_unique_timeseries_role(db: Session, role: int, contact_id: int, status: int): 

557 db_object = db.query(models.TimeseriesRole).filter(models.TimeseriesRole.role == role) \ 

558 .filter(models.TimeseriesRole.contact_id == contact_id) \ 

559 .filter(models.TimeseriesRole.status == status) \ 

560 .first() 

561 return db_object 

562 

563 

564def get_role_id_from_string(db: Session, role_string: str): 

565 sql_command = f""" 

566 SELECT distinct(r.id) FROM timeseries_roles r, 

567 contacts c, 

568 organisations o, 

569 persons p 

570 WHERE (o.longname IN ('{role_string}') OR 

571 o.name IN ('{role_string}') OR 

572 o.city IN ('{role_string}') OR 

573 o.homepage IN ('{role_string}') OR 

574 o.contact_url IN ('{role_string}') OR 

575 p.email IN ('{role_string}') OR 

576 p.name IN ('{role_string}') OR 

577 p.orcid IN ('{role_string}')) AND 

578 r.contact_id=c.id AND 

579 c.person_id=p.id AND 

580 c.organisation_id=o.id 

581 """ 

582 resultproxy = db.execute(sql_command) 

583 id_list = [ rowproxy._asdict()['id'] for rowproxy in resultproxy ] 

584 return id_list 

585 

586# is this internal, or should this also go to public REST api? 

587def get_unique_timeseries_programme(db: Session, name: str, homepage: str): 

588 db_object = db.query(models.TimeseriesProgramme).filter(models.TimeseriesProgramme.name == name) \ 

589 .filter(models.TimeseriesProgramme.homepage == homepage) \ 

590 .first() 

591 return db_object 

592 

593 

594# is this internal, or should this also go to public REST api? 

595def get_unique_timeseries_annotation(db: Session, text: str, contributor_id: int): 

596 db_object = db.query(models.TimeseriesAnnotation).filter(models.TimeseriesAnnotation.text == text) \ 

597 .filter(models.TimeseriesAnnotation.contributor_id == contributor_id) \ 

598 .first() 

599 return db_object 

600 

601 

602def get_contributors_string(programmes, roles): 

603 # sort every section alphabetically and have sub-section titles 

604 # programmes are already unique, but they still need to be sorted 

605 resultp = "programmes: " + ";".join(sorted([programme.longname for programme in programmes])) if len(programmes) > 0 else '' 

606 # organisations might contain duplicates 

607 organisations = set() 

608 [ organisations.add(role.contact.organisation.longname) for role in roles ] 

609 # eliminate dummy organisation 

610 organisations.discard('') 

611 resulto = "organisations: " + ";".join(sorted(organisations)) if len(organisations) > 0 else '' 

612 # persons might contain duplicates 

613 persons = set() 

614 [ persons.add(role.contact.person.name) for role in roles ] 

615 # eliminate dummy person 

616 persons.discard('') 

617 resultc = "persons:" + ";".join(sorted(persons)) if len(persons) > 0 else '' 

618 result = resultp if resultp != '' else '' 

619 if resulto != '': 

620 result = resulto if result == '' else result + '; ' + resulto 

621 if resultc != '': 

622 result = resultc if result == '' else result + '; ' + resultc 

623 return result 

624 

625 

626def get_contributors_list(db: Session, timeseries_ids, format: str = 'text'): 

627 # get time series' programmes 

628 # join(models.Timeseries, Timeseries.programme_id == TimeseriesProgramme.id) is implicit given due to the foreign key 

629 programmes = db.query(models.TimeseriesProgramme) \ 

630 .join(models.Timeseries) \ 

631 .filter(models.Timeseries.id.in_(timeseries_ids), models.TimeseriesProgramme.id != 0).all() 

632 # get all time series roles (with duplicates) 

633 roles = db.execute(select([timeseries_timeseries_roles_table]).where(timeseries_timeseries_roles_table.c.timeseries_id.in_(timeseries_ids))) 

634 # eliminate duplicates 

635 role_ids = set() 

636 [ role_ids.add(role.role_id) for role in roles ] 

637 roles = db.query(models.TimeseriesRole).filter(models.TimeseriesRole.id.in_(role_ids)).all() 

638 # return both programmes and roles 

639 if format == 'text': 

640 result = get_contributors_string(programmes, roles) 

641 elif format == 'json': 

642 result = programmes + roles 

643 else: 

644 status_code=400 

645 message=f"not a valid format: {format}" 

646 result = JSONResponse(status_code=status_code, content=message) 

647 return result 

648 

649 

650def get_request_contributors(db: Session, format: str = 'text', input_handle: UploadFile = File(...)): 

651 f = input_handle.file 

652 timeseries_ids = [int(line.strip()) for line in f.readlines()] 

653 return get_contributors_list(db, timeseries_ids, format) 

654 

655 

656def get_registered_request_contributors(db: Session, rid, format: str = 'text'): 

657 try: 

658 timeseries_ids = db.execute(select([s1_contributors_table]).\ 

659 where(s1_contributors_table.c.request_id == rid)).mappings().first()['timeseries_ids'] 

660 return get_contributors_list(db, timeseries_ids, format) 

661 except: 

662 status_code=400 

663 message=f"not a registered request id: {rid}" 

664 return JSONResponse(status_code=status_code, content=message) 

665 

666 

667def register_request_contributors(db: Session, rid, ids): 

668 try: 

669 db.execute(insert(s1_contributors_table).values(request_id=rid, timeseries_ids=ids)) 

670 db.commit() 

671 status_code = 200 

672 message=f'{rid} successfully registered.' 

673 except IntegrityError as e: 

674 error_code = e.orig.pgcode 

675 if error_code == "23505": 

676 status_code = 443 

677 message=f'{rid} already registered.' 

678 else: 

679 status_code = 442 

680 message=f'database error: error_code' 

681 result = JSONResponse(status_code=status_code, content=message) 

682 return result 

683 

684 

685# is this internal, or should this also go to public REST api? 

686def get_timeseries_roles(db: Session, timeseries_id: int): 

687 return db.execute(select([timeseries_timeseries_roles_table]).where(timeseries_timeseries_roles_table.c.timeseries_id == timeseries_id)) 

688 

689 

690# is this internal, or should this also go to public REST api? 

691def get_timeseries_role_by_id(db: Session, role_id): 

692 return db.query(models.TimeseriesRole).filter(models.TimeseriesRole.id == role_id).first() 

693 

694 

695# is this internal, or should this also go to public REST api? 

696def get_timeseries_annotations(db: Session, timeseries_id: int): 

697 return db.execute(select([timeseries_timeseries_annotations_table]).where(timeseries_timeseries_annotations_table.c.timeseries_id == timeseries_id)) 

698 

699 

700# is this internal, or should this also go to public REST api? 

701def get_timeseries_annotation_by_id(db: Session, annotation_id): 

702 return db.query(models.TimeseriesAnnotation).filter(models.TimeseriesAnnotation.id == annotation_id).first() 

703 

704 

705def list_coverage(db, 

706 station_id: int = None, 

707 variable_id: int = None, 

708 timeseries_id: int = None, 

709 year: int = None, 

710 year_min: int = None, 

711 year_max: int = None, 

712 offset: int = 0, 

713 limit: int = 10, 

714): 

715 """Return rows with optional filtering and pagination.""" 

716 q = db.query(models.YearlyCoverage) 

717 

718 if station_id is not None: 

719 q = q.filter(models.YearlyCoverage.station_id == station_id) 

720 if variable_id is not None: 

721 q = q.filter(models.YearlyCoverage.variable_id == variable_id) 

722 if timeseries_id is not None: 

723 q = q.filter(models.YearlyCoverage.timeseries_id == timeseries_id) 

724 if year is not None: 

725 q = q.filter(models.YearlyCoverage.year == year) 

726 if year_min is not None: 

727 q = q.filter(models.YearlyCoverage.year >= year_min) 

728 if year_max is not None: 

729 q = q.filter(models.YearlyCoverage.year <= year_max) 

730 

731 return q.offset(offset).limit(limit).all() 

732 

733 

734def create_timeseries(db: Session, timeseries: TimeseriesCreate, author_id: int):