Coverage for toardb / utils / utils.py: 82%
313 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 12:51 +0000
1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH
2# SPDX-License-Identifier: MIT
4"""
5Helper functions for TOAR database
7"""
8from sqlalchemy import Table, and_
9from sqlalchemy.orm import Session
10from sqlalchemy.inspection import inspect
11from sqlalchemy.dialects import postgresql
12from fastapi import HTTPException, Request, Header, Depends
13from starlette.datastructures import QueryParams
14from collections import namedtuple
15from copy import copy
16import requests
17import datetime as dt
18from typing import List
20from toardb.utils.settings import base_geopeas_url, userinfo_endpoint
21from toardb.utils.deployment_settings import dashboard_token
23# the following statement only if not in testing (pytest) mode!
24from toardb.utils.database import get_db
25from toardb.contacts.models import Contact, Organisation, Person
26from toardb.timeseries.models import Timeseries, TimeseriesRole, timeseries_timeseries_roles_table
27from toardb.stationmeta.models import StationmetaCore, StationmetaGlobal
28from toardb.data.models import Data
29from toardb.variables.models import Variable
30from toardb.auth_user.models import AuthUser
31from toardb.auth_user.crud import get_eduperson_and_roles, determine_increments
32import toardb
34roles_params = {column.name for column in inspect(TimeseriesRole).c} - {"id"}
37def get_access_rights(request: Request, access_right: str = 'admin', incr: List[int] = [0, 0], db: Session = None):
38 # Do not use underscores; they are not valid in header attributes!
39 user_name = ''
40 user_email = ''
41 auth_user_id = -1
42 role = 'anonymous'
43 personinfo = get_eduperson_and_roles(request=request, db=db, DoIncr=incr)
44 status_code = personinfo['status_code']
45 if status_code != 401:
46 ltoken = True
47 userinfo = personinfo['userinfo']
48 role = personinfo['role']
49 if ("eduperson_entitlement" in userinfo and \
50 f"urn:geant:helmholtz.de:res:toar-data{access_right}#login.helmholtz.de" \
51 in userinfo["eduperson_entitlement"]) or access_right in [":data-download", ":map-data-download"]:
52 user_name = userinfo["name"]
53 user_email = userinfo["email"]
54 db_user = db.query(AuthUser).filter(AuthUser.email == user_email).first()
55 if db_user:
56 auth_user_id = db_user.id
57 else:
58 status_code = 401
59 if access_right == ":data-download" and (personinfo["max_timeseries"] is not None):
60 if personinfo["num_timeseries"] > personinfo["max_timeseries"]:
61 status_code = 401
62 if access_right == ":map-data-download" and (personinfo["max_gridded"] is not None):
63 if personinfo["num_gridded"] > personinfo["max_gridded"]:
64 status_code = 401
65 else:
66 role = 'unauthorized'
67 access_dict = { "status_code": status_code,
68 "user_name": user_name,
69 "user_email": user_email,
70 "auth_user_id": auth_user_id,
71 "role": role }
72 return access_dict
75def get_admin_access_rights(request: Request, db: Session = Depends(get_db)):
76 return get_access_rights(request, ':admin', db=db)
79def get_station_md_change_access_rights(request: Request, db: Session = Depends(get_db)):
80 return get_access_rights(request, ':station-md-change', db=db)
83def get_timeseries_md_change_access_rights(request: Request, db: Session = Depends(get_db)):
84 return get_access_rights(request, ':timeseries-md-change', db=db)
87def get_data_change_access_rights(request: Request, db: Session = Depends(get_db)):
88 return get_access_rights(request, ':data-change', db=db)
91def get_register_contributors_access_rights(request: Request, db: Session = Depends(get_db)):
92 return get_access_rights(request, ':contributors-register', db=db)
95def get_data_download_access_rights(request: Request, db: Session = Depends(get_db)):
96 # there is only need for (repeated) authorization via AAI, if request is not coming from the dashboard or any other service
97 lfromdashboard = request.headers.get('DashboardToken') == dashboard_token
98 if not lfromdashboard:
99 access = get_access_rights(request, ':data-download', [1, 1], db=db)
100 else:
101 access = {}
102 access['lfromdashboard'] = lfromdashboard
103 return access
106def get_map_data_download_access_rights(request: Request, db: Session = Depends(get_db)):
107 # there is only need for (repeated) authorization via AAI, if request is not coming from the dashboard or any other service
108 lfromdashboard = request.headers.get('DashboardToken') == dashboard_token
109 if not lfromdashboard:
110 # the increment depends on the number of touched time series
111 incr = determine_increments(request, db)
112 access = get_access_rights(request, ':map-data-download', [2, incr['num_gridded']], db=db)
113 else:
114 access = {}
115 access['lfromdashboard'] = lfromdashboard
116 return access
119# function to return code for given value
120def get_str_from_value(enum_dict, value) -> str:
121 return tuple(filter(lambda x: x.value == value, enum_dict))[0].string
123def get_displaystr_from_value(enum_dict, value) -> str:
124 return tuple(filter(lambda x: x.value == value, enum_dict))[0].display_str
126# function to return value for given code
127def get_value_from_str(enum_dict, string) -> int:
128 try:
129 return tuple(filter(lambda x: x.string == string, enum_dict))[0].value
130 except:
131 raise ValueError(f"value not known: {string}")
133# function to return value for given display string
134def get_value_from_display_str(enum_dict, string) -> int:
135 return tuple(filter(lambda x: x.display_str == string, enum_dict))[0].value
138# get human readable fields, if database field is controlled (by vocabulary)
139def get_hr_value(table_str,field,value):
140 index = None
141 for mid in ['core', 'global', 'glob', 'annotations', 'roles', 'changelog']:
142 if table_str + f'_{mid}_' + field in toardb.toardb.controlled_fields:
143 index = table_str + f'_{mid}_' + field
144 if index:
145 vocabulary = getattr(toardb.toardb, toardb.toardb.controlled_fields[index])
146 value = get_str_from_value(vocabulary,int(value))
147 return value
150# translate filters that contain controlled vocabulary
151def translate_convoc_list(values, table, display_name):
152 try:
153 return [get_value_from_str(table,v) for v in values]
154 except ValueError:
155 raise HTTPException(status_code=470, detail=f"{display_name} not known: {values}")
158# expand subdicts except for additional_metadata
159def normalize_metadata(metadata):
160 normalized_metadata = {}
161 for key, val in metadata.items():
162 if isinstance(val, dict) and val and key != "additional_metadata":
163 normalized_metadata.update(
164 {
165 f"{key}_{sub_key}": sub_val
166 for sub_key, sub_val in normalize_metadata(val).items()
167 }
168 )
169 else:
170 normalized_metadata[key] = val
171 return normalized_metadata
174def pop_non_merged(query_params, to_remove=["daterange", "flags"]):
175 items = list(query_params.multi_items())
176 filtered_items = [(k, v) for k, v in items if k not in to_remove]
177 return QueryParams(filtered_items)
179#
180def create_filter(qps, endpoint):
182 # for ideas on how to create filter on special roles see:
183 # https://gitlab.jsc.fz-juelich.de/esde/toar-data/toardb_fastapi/-/issues/95#note_144292
185 # determine allowed query parameters (first *only* from Timeseries)
186 timeseries_params = {column.name for column in inspect(Timeseries).c} | {"has_role"}
187 timeseries_params = timeseries_params | {"additional_metadata-"}
188 gis_params = {"bounding_box", "altitude_range"}
189 if endpoint in ['search', 'timeseries']:
190 core_params = {column.name for column in inspect(StationmetaCore).c if column.name not in ['id']}
191 else:
192 core_params = {column.name for column in inspect(StationmetaCore).c}
193 core_params |= {"globalmeta", "station_additional_metadata-"}
194 global_params = {column.name for column in inspect(StationmetaGlobal).c if column.name not in ['id','station_id']}
195 data_params = {column.name for column in inspect(Data).c} | {"daterange", "format"}
196 timeseries_merged_params = data_params | {"station_code", "variable_id", "id", "has_role"}
197 ambig_params = {"station_id", "station_changelog", "station_country", "station_additional_metadata"}
198 variable_params = {column.name for column in inspect(Variable).c}
199 person_params = {column.name for column in inspect(Person).c}
200 allrel_params = {"limit", "offset", "fields", "format"}
201 profiling_params = {"profile", "profile_format", "timing"}
203 # pagination
204 offset= int(qps.get("offset", 0))
205 try:
206 limit = int(qps.get("limit", 10))
207 except:
208 limit = qps.get("limit")
209 if limit == "None":
210 limit = None
211 else:
212 raise ValueError(f"Wrong value for limit given: {limit}")
214 # fields and format are no filter options
215 fields = qps.get("fields", "")
216 format = qps.get("format", 'json')
218 allowed_params = allrel_params.copy()
219 allowed_params |= profiling_params
220 if endpoint in {'stationmeta'}:
221 allowed_params |= gis_params | core_params | global_params
222 elif endpoint in {'timeseries'}:
223 allowed_params |= timeseries_params | roles_params
224 elif endpoint in {'search'}:
225 allowed_params |= gis_params | core_params | global_params | timeseries_params | roles_params | ambig_params
226 elif endpoint in {'data'}:
227 allowed_params |= data_params | profiling_params
228 elif endpoint in {'timeseries_merged'}:
229 allowed_params |= timeseries_merged_params | profiling_params
230 elif endpoint in {'variables'}:
231 allowed_params |= variable_params
232 elif endpoint in {'persons'}:
233 allowed_params |= person_params
234 else:
235 raise ValueError(f"Wrong endpoint given: {endpoint}")
237 query_params = qps
238 if endpoint == 'timeseries_merged':
239 query_params = pop_non_merged(qps)
241 if fields:
242 for field in fields.split(','):
243 if field not in allowed_params:
244 raise ValueError(f"Wrong field given: {field}")
245 if fields.find("globalmeta") >= 0:
246 fields = fields.replace("globalmeta","")
247 fields += ','.join(global_params)
249 t_filter = []
250 t_r_filter = []
251 s_c_filter = []
252 s_g_filter = []
253 d_filter = []
254 v_filter = []
255 p_filter = []
256 # query_params is a multi-dict!
257 for param_long in query_params:
258 param = param_long.split('>')[0]
259 if param not in allowed_params: #inform user, that an unknown parameter name was used (this could be a typo and falsify the result!)
260 raise KeyError(f"An unknown argument was received: {param}.")
261 if param in allrel_params or param in profiling_params:
262 continue
263 if param == 'station_code':
264 param = 'codes'
265 values = [item.strip() for v in query_params.getlist(param_long) for item in v.split(',')]
266 # make sure ids are ints
267 if param.endswith("id"):
268 try:
269 int_values = [ int(v) for v in values ]
270 values = int_values
271 except:
272 raise ValueError(f"Wrong value (not int) given: {param}")
273 # allow '+' in datestring (for adding timezone information)
274 if param.endswith("date"):
275 try:
276 value = dt.datetime.fromisoformat(values[0])
277 except:
278 raise ValueError(f"Wrong value for time given: {values[0]}")
279 # request package transforms blank to '+' --> return to blank
280 try:
281 values = [ value.replace('+',' ') for value in values ]
282 except:
283 pass
284 if endpoint in ["stationmeta", "timeseries", "timeseries_merged", "search"] and param in core_params:
285 #check for parameters of the controlled vocabulary
286 if param == "timezone":
287 values = translate_convoc_list(values, toardb.toardb.TZ_vocabulary, "timezone")
288 elif param == "coordinate_validation_status":
289 values = translate_convoc_list(values, toardb.toardb.CV_vocabulary, "coordinate validation status")
290 elif param == "country":
291 values = translate_convoc_list(values, toardb.toardb.CN_vocabulary, "country")
292 elif param == "type":
293 values = translate_convoc_list(values, toardb.toardb.ST_vocabulary, "type")
294 elif param == "type_of_area":
295 values = translate_convoc_list(values, toardb.toardb.TA_vocabulary, "type of area")
296 elif param == "station_additional_metadata-":
297 param = f"{param_long[8:]}"
298 # exceptions for special fields (codes, name)
299 if param == 'codes':
300 tmp_filter = []
301 for v in values:
302 tmp_filter.append(f"('{v}'=ANY(stationmeta_core.codes))")
303 tmp_filter = " OR ".join(tmp_filter)
304 s_c_filter.append(f"({tmp_filter})")
305 elif param == 'name':
306 s_c_filter.append(f"LOWER(stationmeta_core.name) LIKE '%{values[0].lower()}%'")
307 elif param_long.split('>')[0] == "station_additional_metadata-":
308 val_mod = [ f"'\"{val}\"'::text" for val in values ]
309 values = ",".join(val_mod)
310 s_c_filter.append(f"to_json(stationmeta_core.{param})::text IN ({values})")
311 else:
312 s_c_filter.append(f"stationmeta_core.{param} IN {values}")
313 elif endpoint in ["stationmeta", "search"] and param in global_params:
314 if param == "climatic_zone_year2016":
315 values = translate_convoc_list(values, toardb.toardb.CZ_vocabulary, "climatic zone year2016")
316 elif param == "toar1_category":
317 values = translate_convoc_list(values, toardb.toardb.TC_vocabulary, "TOAR-I category")
318 elif param == "toar2_category":
319 values = translate_convoc_list(values, toardb.toardb.TA_vocabulary, "TOAR-II category")
320 elif param == "htap_region_tier1_year2010":
321 values = translate_convoc_list(values, toardb.toardb.TR_vocabulary, "HTAP region TIER1 year2010")
322 elif param == "dominant_landcover_year2012":
323 values = translate_convoc_list(values, toardb.toardb.LC_vocabulary, "landcover type")
324 elif param == "dominant_ecoregion_year2017":
325 values = translate_convoc_list(values, toardb.toardb.ER_vocabulary, "ECO region type")
326 s_g_filter.append(f"stationmeta_global.{param} IN {values}")
327 elif endpoint in ["stationmeta", "search"] and param in gis_params:
328 if param == "bounding_box":
329 min_lat, min_lon, max_lat, max_lon = values
330 bbox= f'SRID=4326;POLYGON (({min_lon} {min_lat}, {min_lon} {max_lat}, {max_lon} {max_lat}, {max_lon} {min_lat}, {min_lon} {min_lat}))'
331 s_c_filter.append(f"ST_CONTAINS(ST_GeomFromEWKT('{bbox}'), coordinates)")
332 else:
333 s_c_filter.append(f"ST_Z(coordinates) BETWEEN {values[0]} AND {values[1]}")
334 elif endpoint in ["timeseries", "timeseries_merged", "search"]:
335 #check for parameters of the controlled vocabulary
336 if param == "sampling_frequency":
337 values = translate_convoc_list(values, toardb.toardb.SF_vocabulary, "sampling_frequency")
338 elif param == "aggregation":
339 values = translate_convoc_list(values, toardb.toardb.AT_vocabulary, "aggregation")
340 elif param == "data_origin_type":
341 values = translate_convoc_list(values, toardb.toardb.OT_vocabulary, "data origin type")
342 elif param == "data_origin":
343 values = translate_convoc_list(values, toardb.toardb.DO_vocabulary, "data origin")
344 elif param == "additional_metadata-":
345 param = param_long
346 if param == "additional_metadata->'absorption_cross_section'":
347 trlist = translate_convoc_list(values, toardb.toardb.CS_vocabulary, "absorption_cross_section")
348 values = [ str(val) for val in trlist ]
349 param = f"timeseries.{param}"
350 elif param == "additional_metadata->'sampling_type'":
351 trlist = translate_convoc_list(values, toardb.toardb.KS_vocabulary, "sampling_type")
352 values = [ str(val) for val in trlist ]
353 param = f"timeseries.{param}"
354 elif param == "additional_metadata->'calibration_type'":
355 trlist = translate_convoc_list(values, toardb.toardb.CT_vocabulary, "calibration_type")
356 values = [ str(val) for val in trlist ]
357 param = f"timeseries.{param}"
358 else:
359 val_mod = [ f"'\"{val}\"'::text" for val in values ]
360 values = "(" + ",".join(val_mod) + ")"
361 param = f"to_json(timeseries.{param})::text"
362 if param == "has_role":
363 operator = "IN"
364 join_operator = "OR"
365 if (values[0][0] == '~'):
366 operator = "NOT IN"
367 join_operator = "AND"
368 values[0] = values[0][1:]
369 t_r_filter.append(f"organisations.longname {operator} {values}")
370 t_r_filter.append(f"organisations.name {operator} {values}")
371 t_r_filter.append(f"organisations.city {operator} {values}")
372 t_r_filter.append(f"organisations.homepage {operator} {values}")
373 t_r_filter.append(f"organisations.contact_url {operator} {values}")
374 t_r_filter.append(f"persons.email {operator} {values}")
375 t_r_filter.append(f"persons.name {operator} {values}")
376 t_r_filter.append(f"persons.orcid {operator} {values}")
377 t_r_filter = f" {join_operator} ".join(t_r_filter)
378 elif param_long.split('>')[0] == "additional_metadata-":
379 t_filter.append(f"{param} IN {values}")
380 else:
381 t_filter.append(f"timeseries.{param} IN {values}")
382 elif param in data_params:
383 if param == "daterange":
384 start_date = dt.datetime.fromisoformat(values[0])
385 stop_date = dt.datetime.fromisoformat(values[1])
386 d_filter.append(f"datetime BETWEEN '{start_date}' AND '{stop_date}'")
387 elif param in ("flags", "format"):
388 # translation of flags should be done in data.crud
389 continue
390 else:
391 d_filter.append(f"data.{param} IN {values}")
392 elif endpoint == "variables":
393 v_filter.append(f"variables.{param} IN {values}")
394 elif param in person_params:
395 p_filter.append(f"persons.{param} IN {values}")
398 t_filter = " AND ".join(t_filter).replace('[','(').replace(']',')')
399 t_r_filter = '(' + "".join(t_r_filter).replace('[','(').replace(']',')') + ')'
400 if t_r_filter == '()':
401 t_r_filter = ''
402 s_c_filter = " AND ".join(s_c_filter).replace('[','(').replace(']',')')
403 s_g_filter = " AND ".join(s_g_filter).replace('[','(').replace(']',')')
404 d_filter = " AND ".join(d_filter).replace('[','(').replace(']',')')
405 v_filter = " AND ".join(v_filter).replace('[','(').replace(']',')')
406 p_filter = " AND ".join(p_filter).replace('[','(').replace(']',')')
407 filters = {
408 "t_filter": t_filter,
409 "t_r_filter":t_r_filter,
410 "s_c_filter": s_c_filter,
411 "s_g_filter": s_g_filter,
412 "d_filter": d_filter,
413 "v_filter": v_filter,
414 "p_filter": p_filter,
415 }
416 return limit, offset, fields, format, filters
419###
420# Rasdaman does not run stable!
421# --> since GEO PEAS runs in DEBUG mode, its messages flood the error log file!
422#
423# only activate the following lines if you want to update pages!
424###
425# also get provenance information
426geopeas_services = [ 'topography_srtm', 'ecoregion', 'stable_nightlights',
427 'climatic_zone', 'nox_emissions', 'landcover',
428 'major_road', 'population_density', 'htap_region_tier1' ]
429provenance = {}
430#for service in geopeas_services:
431# result = requests.get(f"{base_geopeas_url}/{service}/").json()
432# provenance[service] = result['provenance']
433# # major_road has different dict entries
434# tmp_provenance = copy(provenance[service])
435# # htap_region_tier1 does not provide a dict
436# try:
437# keys = [x for x in result['provenance'].keys()]
438# for key in keys:
439# if key not in ['units', 'data_source', 'citation', 'doi']:
440# result['provenance'].pop(key)
441# if len(result['provenance']) == 0:
442# provenance[service] = tmp_provenance
443# except:
444# continue
445for service in geopeas_services:
446 provenance[service] = "dummy text"