Coverage for toardb/stationmeta/crud.py: 82%
464 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-03 20:32 +0000
1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH
2# SPDX-License-Identifier: MIT
4"""
5Create, Read, Update, Delete functionality
7"""
9import json
10import requests
12from typing import List
13from pydantic import ValidationError
14from geoalchemy2.shape import to_shape
15from geoalchemy2.types import Geometry
16from geoalchemy2.elements import WKBElement, WKTElement
17from sqlalchemy import cast, Text, insert, update, delete, select, and_, String, text
18from sqlalchemy.orm import Session
19from sqlalchemy.engine import Engine
20from sqlalchemy.dialects.postgresql import JSONB, ARRAY
21from sqlalchemy.inspection import inspect
22from fastapi import File, UploadFile
23from fastapi.responses import JSONResponse
25from . import models
26from .models import StationmetaCore, StationmetaGlobal, StationmetaChangelog, \
27 stationmeta_core_stationmeta_roles_table, stationmeta_core_stationmeta_annotations_table
28from .schemas import get_coordinates_from_geom, get_geom_from_coordinates, get_abbreviation_from_code_description, get_code_description_from_abbreviation, \
29 Stationmeta, StationmetaCreate, StationmetaPatch, StationmetaGlobalFields, Coordinates, get_coordinates_from_string
30from toardb.utils.utils import get_value_from_str, get_str_from_value, get_hr_value, create_filter
31from toardb.utils.settings import base_geopeas_url
32import toardb
35def get_stationmeta_core(db: Session, station_code: str):
36 db_object = db.query(models.StationmetaCore).filter(models.StationmetaCore.codes.any(station_code.strip())).first()
37 # there is a mismatch with coordinates and additional_metadata
38 if db_object:
39 if isinstance(db_object.coordinates, (WKBElement, WKTElement)):
40 db_object.coordinates = get_coordinates_from_geom(db_object.coordinates)
41 db_object.additional_metadata = str(db_object.additional_metadata).replace("'",'"')
42 return db_object
45def get_stationmeta(db: Session, station_code: str, fields: str = None):
46 if fields:
47 global_params = {column.name for column in inspect(StationmetaGlobal).c if column.name not in ['id','station_id']}
48 lconstr_glob = any(field in global_params for field in fields.split(','))
49 if fields.find("globalmeta") >= 0:
50 fields = fields.replace("globalmeta","")
51 fields += ','.join(global_params)
52 lconstr_glob = True
53 fields2 = []
54 for field in fields.split(','):
55 if field == "id":
56 fields2.append("stationmeta_core.id")
57 elif field == "coordinates":
58 fields2.append("ST_AsText(coordinates)")
59 else:
60 fields2.append(field)
61 tmp_object = db.query(*map(text,fields2)). \
62 select_from(StationmetaCore).filter(models.StationmetaCore.codes.any(station_code.strip())). \
63 join(StationmetaGlobal).filter(models.StationmetaCore.id == models.StationmetaGlobal.station_id). \
64 first()
65 if tmp_object:
66 db_object = dict(zip(fields.split(','),tmp_object))
67 else:
68 return tmp_object
69 # there is a mismatch with coordinates and additional_metadata
70 if lconstr_glob:
71 db_object_cp = db_object.copy()
72 global_atts = {}
73 for key in db_object:
74 if key in global_params:
75 global_atts[key] = db_object[key]
76 del db_object_cp[key]
77 db_object = db_object_cp
78 db_object["globalmeta"] = StationmetaGlobalFields(**global_atts)
79 try:
80 db_object['coordinates'] = get_coordinates_from_string(db_object['coordinates'])
81 except:
82 pass
83 try:
84 db_object['additional_metadata'] = str(db_object['additional_metadata']).replace("'",'"')
85 except:
86 pass
87 else:
88 db_object = db.query(models.StationmetaCore).filter(models.StationmetaCore.codes.any(station_code.strip())).first()
89 # there is a mismatch with coordinates and additional_metadata
90 if db_object:
91 try:
92 if isinstance(db_object.coordinates, (WKBElement, WKTElement)):
93 db_object.coordinates = get_coordinates_from_geom(db_object.coordinates)
94 except:
95 pass
96 try:
97 db_object.additional_metadata = str(db_object.additional_metadata).replace("'",'"')
98 except:
99 pass
100 return db_object
103def get_stationmeta_by_id(db: Session, station_id: int):
104 db_object = db.query(models.StationmetaCore).filter(models.StationmetaCore.id == station_id).first()
105 # there is a mismatch with coordinates and additional_metadata
106 if db_object:
107 if isinstance(db_object.coordinates, (WKBElement, WKTElement)):
108 db_object.coordinates = get_coordinates_from_geom(db_object.coordinates)
109 db_object.additional_metadata = str(db_object.additional_metadata).replace("'",'"')
110 return db_object
113def station_id_exists(db: Session, station_id: int):
114 db_object = db.query(models.StationmetaCore).filter(models.StationmetaCore.id == station_id).first()
115 return (db_object != None)
118def get_all_stationmeta(db: Session, path_params, query_params):
119 try:
120 limit, offset, fields, format, filters = create_filter(query_params, "stationmeta")
121 s_c_filter = filters["s_c_filter"]
122 s_g_filter = filters["s_g_filter"]
123 except (KeyError, ValueError) as e:
124 status_code=400
125 return JSONResponse(status_code=status_code, content=str(e))
126 global_params = {column.name for column in inspect(StationmetaGlobal).c if column.name not in ['id','station_id']}
128 if fields:
129 fields2 = []
130 for field in fields.split(','):
131 if field == "id":
132 fields2.append("stationmeta_core.id")
133 elif field == "coordinates":
134 fields2.append("ST_AsText(coordinates)")
135 else:
136 fields2.append(field)
137 db_objects_l = db.query(*map(text,fields2)).select_from(StationmetaCore,StationmetaGlobal).filter(text(s_c_filter)). \
138 join(StationmetaGlobal).filter(and_(StationmetaCore.id == StationmetaGlobal.station_id, text(s_g_filter))). \
139 distinct().order_by(models.StationmetaCore.id). \
140 limit(limit).offset(offset)
141 db_objects = []
142 for db_object_immut in db_objects_l:
143 db_object = dict(zip(fields.split(','),db_object_immut))
144 # there is a mismatch with coordinates and additional_metadata
145 try:
146 db_object['coordinates'] = get_coordinates_from_string(db_object['coordinates'])
147 except:
148 pass
149 try:
150 db_object['additional_metadata'] = str(db_object['additional_metadata']).replace("'",'"')
151 except:
152 pass
153 db_object_cp = db_object.copy()
154 global_atts = {}
155 for key in db_object:
156 if key in global_params:
157 global_atts[key] = db_object[key]
158 del db_object_cp[key]
159 db_object = db_object_cp
160 if global_atts:
161 db_object["globalmeta"] = StationmetaGlobalFields(**global_atts)
162 db_objects.append(db_object)
163 else:
164 db_objects = db.query(models.StationmetaCore).filter(text(s_c_filter)). \
165 join(StationmetaGlobal).filter(and_(StationmetaCore.id == StationmetaGlobal.station_id, text(s_g_filter))). \
166 order_by(models.StationmetaCore.id). \
167 limit(limit).offset(offset).all()
169 for db_object in db_objects:
170 # there is a mismatch with coordinates and additional_metadata
171 if isinstance(db_object.coordinates, (WKBElement, WKTElement)):
172 db_object.coordinates = get_coordinates_from_geom(db_object.coordinates)
173 db_object.additional_metadata = str(db_object.additional_metadata).replace("'",'"')
175 if limit:
176 return db_objects[:limit]
177 else:
178 return db_objects
181# is this internal, or should this also go to public REST api?
182# - get_stationmeta_roles
183# - get_stationmeta_changelog
184# - get_unique_stationmeta_role
185# - get_stationmeta_role_by_id
186# - get_unique_stationmeta_annotation
187def get_stationmeta_roles(db: Session, station_id: int):
188 return db.execute(select([stationmeta_core_stationmeta_roles_table]).where(stationmeta_core_stationmeta_roles_table.c.station_id == station_id))
191def get_stationmeta_changelog(db: Session, station_id: int):
192 return db.query(models.StationmetaChangelog).filter(models.StationmetaChangelog.station_id == station_id).order_by(models.StationmetaChangelog.datetime).all()
195def get_unique_stationmeta_role(db: Session, role: int, contact_id: int, status: int):
196 db_object = db.query(models.StationmetaRole).filter(models.StationmetaRole.role == role) \
197 .filter(models.StationmetaRole.contact_id == contact_id) \
198 .filter(models.StationmetaRole.status == status) \
199 .first()
200 return db_object
202def get_stationmeta_role_by_id(db: Session, role_id):
203 return db.query(models.StationmetaRole).filter(models.StationmetaRole.id == role_id).first()
206def get_unique_stationmeta_annotation(db: Session, text: str, contributor_id: int):
207 db_object = db.query(models.StationmetaAnnotation).filter(models.StationmetaAnnotation.text == text) \
208 .filter(models.StationmetaAnnotation.contributor_id == contributor_id) \
209 .first()
210 return db_object
213# is this internal, or should this also go to public REST api?
214def get_stationmeta_annotations(db: Session, station_id: int):
215 return db.execute(select([stationmeta_core_stationmeta_annotations_table]).where(stationmeta_core_stationmeta_annotations_table.c.station_id == station_id))
218def get_stationmeta_global(db: Session, station_id: int):
219 db_object = db.query(models.StationmetaGlobal).filter(models.StationmetaGlobal.station_id == station_id) \
220 .first()
221 return db_object
224def determine_stationmeta_global(db, tmp_coordinates, country):
225 # GEO-PEAS expects country to be a two-letter-string
226 country = get_str_from_value(toardb.toardb.CN_vocabulary,country)
227 db_globalservices = db.query(models.StationmetaGlobalService).all()
228 globalmeta_dict = {}
229 for db_service in db_globalservices:
230 request=db_service.service_url.format(base_url=f'{base_geopeas_url}',lat=tmp_coordinates["lat"],lon=tmp_coordinates["lng"],
231 country=country)
232 try:
233 result = requests.get(request).json()
234 # htap_region_tier1 (also major_road) does not return "agg_function"!
235 try:
236 key = result["properties"]["agg_function"]
237 except:
238 if db_service.variable_name != 'distance_to_major_road_year2020':
239 key = "value"
240 else:
241 key = 'distance'
242 try:
243 value = result["properties"][key]
244 except:
245 value = result[key]
246 except:
247 continue
248 globalmeta_dict[db_service.variable_name] = value
249 return globalmeta_dict
252def create_stationmeta(db: Session, engine: Engine, stationmeta: StationmetaCreate,
253 author_id: int, force: bool):
254 stationmeta_dict = stationmeta.dict()
255 roles_data = stationmeta_dict.pop('roles', None)
256 annotations_data = stationmeta_dict.pop('annotations', None)
257 aux_images_data = stationmeta_dict.pop('aux_images', None)
258 aux_docs_data = stationmeta_dict.pop('aux_docs', None)
259 aux_urls_data = stationmeta_dict.pop('aux_urls', None)
260 db_stationmeta = models.StationmetaCore(**stationmeta_dict)
261 # there's a mismatch with coordinates --> how to automatically switch back and forth?!
262 tmp_coordinates = db_stationmeta.coordinates
263 db_stationmeta.coordinates = get_geom_from_coordinates(Coordinates(**db_stationmeta.coordinates))
264 # there's also a mismatch with additional_metadata --> BUT: this should not be switched back!
265 # in upload command, we have now: "additional_metadata": "{}"
266 # but return from this method gives: "additional_metadata": {}
267 # ==> there is a mismatch between model(JSONB) and schema(JSON)
268 db_stationmeta.additional_metadata = json.loads(str(db_stationmeta.additional_metadata).replace("'",'"'))
269 num_records = 0
270 if not force:
271 with engine.begin() as connection:
272 radius = 10
273 db_cmd = f"select id,codes,name,coordinates,state,country from stationmeta_core where ST_DistanceSphere(stationmeta_core.coordinates, '{db_stationmeta.coordinates}') < {radius}"
274 records = connection.execute(db_cmd)
275 num_records = records.rowcount
276 if num_records > 0:
277 status_code=444
278 if num_records == 1:
279 message={"message": f'already in database --> patch record (add station code)'}
280 else:
281 message = {"message": 'more than one station falls within the given radius!\n' + \
282 'choose which station record to patch (add station code)'}
283 station_ids = []
284 station_records = []
285 for record in records:
286 station_ids.append(record[0])
287 station_records.append((*record[1:2],str(get_coordinates_from_geom(WKBElement(record[3]))),*record[4:]))
288 message["station_ids"] = station_ids
289 message["station_records"] = station_records
290 engine.dispose()
291 if num_records == 0:
292 status_code=200
293 db_stationmeta.country = get_value_from_str(toardb.toardb.CN_vocabulary,db_stationmeta.country)
294 db_stationmeta.timezone = get_value_from_str(toardb.toardb.TZ_vocabulary,db_stationmeta.timezone)
295 db_stationmeta.type = get_value_from_str(toardb.toardb.ST_vocabulary,db_stationmeta.type)
296 db_stationmeta.type_of_area = get_value_from_str(toardb.toardb.TA_vocabulary,db_stationmeta.type_of_area)
297 db_stationmeta.coordinate_validation_status = get_value_from_str(toardb.toardb.CV_vocabulary,db_stationmeta.coordinate_validation_status)
298 db.add(db_stationmeta)
299 result = db.commit()
300 db.refresh(db_stationmeta)
301 # get stationmeta_core_id
302 stationmeta_core_id = db_stationmeta.id
303 message={"message":f'new station: {db_stationmeta.codes},{db_stationmeta.name},{tmp_coordinates}',
304 "station_id": stationmeta_core_id}
305 # store roles and update association table
306 if roles_data:
307 for r in roles_data:
308 db_role = models.StationmetaRole(**r)
309 db_role.role = get_value_from_str(toardb.toardb.RC_vocabulary,db_role.role)
310 db_role.status = get_value_from_str(toardb.toardb.RS_vocabulary,db_role.status)
311 # check whether role is already present in database
312 db_object = get_unique_stationmeta_role(db, db_role.role, db_role.contact_id, db_role.status)
313 if db_object:
314 role_id = db_object.id
315 else:
316 # Something is going wrong here!
317 # Is the model StationmetaRole correctly defined?!
318 del db_role.contact
319 db.add(db_role)
320 db.commit()
321 db.refresh(db_role)
322 role_id = db_role.id
323 db.execute(insert(stationmeta_core_stationmeta_roles_table).values(station_id=stationmeta_core_id, role_id=role_id))
324 db.commit()
325 # store annotations and update association table
326 if annotations_data:
327 for a in annotations_data:
328 db_annotation = models.StationmetaAnnotation(**a)
329 db_annotation.kind = get_value_from_str(toardb.toardb.AK_vocabulary,db_annotation.kind)
330 # check whether annotation is already present in database
331 db_object = get_unique_stationmeta_annotation(db, db_annotation.text, db_annotation.contributor_id)
332 if db_object:
333 annotation_id = db_object.id
334 else:
335 db.add(db_annotation)
336 db.commit()
337 db.refresh(db_annotation)
338 annotation_id = db_annotation.id
339 db.execute(insert(stationmeta_core_stationmeta_annotations_table).values(station_id=stationmeta_core_id, annotation_id=annotation_id))
340 db.commit()
341 # store aux_images
342 if aux_images_data:
343 for i in aux_images_data:
344 db_aux_image = models.StationmetaAuxImage(**i)
345 db_aux_image.station_id = stationmeta_core_id
346 db.add(db_aux_image)
347 db.commit()
348 db.refresh(db_aux_image)
349 # store aux_docs
350 if aux_docs_data:
351 for d in aux_docs_data:
352 db_aux_doc = models.StationmetaAuxDoc(**d)
353 db_aux_doc.station_id = stationmeta_core_id
354 db.add(db_aux_doc)
355 db.commit()
356 db.refresh(db_aux_doc)
357 # store aux_urls
358 if aux_urls_data:
359 for u in aux_urls_data:
360 db_aux_url = models.StationmetaAuxUrl(**u)
361 db_aux_url.station_id = stationmeta_core_id
362 db.add(db_aux_url)
363 db.commit()
364 db.refresh(db_aux_url)
365 # determine stationmeta_global
366 globalmeta_dict = determine_stationmeta_global(db, tmp_coordinates,db_stationmeta.country)
367 db_globalmeta = models.StationmetaGlobal(**globalmeta_dict)
368 db_globalmeta.station_id = stationmeta_core_id
369 db.add(db_globalmeta)
370 db.commit()
371 db.refresh(db_globalmeta)
372 # create changelog entry
373 type_of_change = get_value_from_str(toardb.toardb.CL_vocabulary,"Created")
374 description="station created"
375 db_changelog = StationmetaChangelog(description=description, station_id=stationmeta_core_id, author_id=author_id, type_of_change=type_of_change,
376 old_value='', new_value='')
377 db.add(db_changelog)
378 db.commit()
379 # there's a mismatch with coordinates --> how to automatically switch back and forth?!
380 db_stationmeta.coordinates = tmp_coordinates
381 res ={}
382 res['detail'] = message
383 return JSONResponse(status_code=status_code, content=res)
386def get_field_from_record(db, station_id, field, db_stationmeta):
387 value = str(getattr(db_stationmeta,field))
388 # if needed, convert this to human readable format (controlled vocabulary)
389 value = get_hr_value('stationmeta',field,value)
390 if isinstance(value, str):
391 value = f"'{value}'"
392 # I forgot to add the code to the following if/else-construct?!
393 if field in ['roles', 'annotations', 'aux_images', 'aux_docs', 'aux_urls', 'globalmeta']:
394 field = f"'{field}': " + value
395 else:
396 field = f"'{field}': " + value
397 return field
400def patch_stationmeta(db: Session, description: str,
401 station_id: int, stationmeta: StationmetaPatch,
402 author_id: int):
404 # still to be done:
405 # collect all old/new_values from different sections!
407 no_log = (description == 'NOLOG')
408 stationmeta_dict = stationmeta.dict()
409 # delete empty fields from stationmeta_dict already at this place, to be able to
410 # distinguish between "single value correction in metadata" and "comprehensive metadata revision"
411 # (see controlled vocabulary "CL_vocabulary")
412 stationmeta_dict2 = {k: v for k, v in stationmeta_dict.items() if v is not None}
413 number_of_elements = len(stationmeta_dict2)
414 if not no_log:
415 if (number_of_elements == 1):
416 type_of_change = get_value_from_str(toardb.toardb.CL_vocabulary,"SingleValue")
417 else:
418 type_of_change = get_value_from_str(toardb.toardb.CL_vocabulary,"Comprehensive")
419 roles_data = stationmeta_dict.pop('roles', None)
420 annotations_data = stationmeta_dict.pop('annotations', None)
421 aux_images_data = stationmeta_dict.pop('aux_images', None)
422 aux_docs_data = stationmeta_dict.pop('aux_docs', None)
423 aux_urls_data = stationmeta_dict.pop('aux_urls', None)
424 globalmeta_data = stationmeta_dict.pop('globalmeta', None)
425 # there's a mismatch with coordinates --> how to automatically switch back and forth?!
426 # ==> the following two commands are not working
427 # ==> workaround
428 # db.query(models.StationmetaCore).filter(models.StationmetaCore.id == stationm_id).update(stationmeta_dict2)
429 db_obj = models.StationmetaCore(**stationmeta_dict)
430 tmp_coordinates = db_obj.coordinates
431 db_stationmeta = db.query(models.StationmetaCore).get(station_id)
432 db_stationmeta.coordinates = get_geom_from_coordinates(db_stationmeta.coordinates)
433 if not no_log:
434 old_value="{"
435 number_of_commas = number_of_elements - 1
436 for k, v in stationmeta_dict.items():
437 if v is not None:
438 # prepare changelog entry
439 if not no_log:
440 field = get_field_from_record(db, station_id, k, db_stationmeta)
441 if (number_of_commas == 0):
442 old_value=old_value + field + '}'
443 else:
444 old_value=old_value + field + ','
445 number_of_commas = number_of_commas - 1
446 setattr(db_stationmeta,k,stationmeta_dict[k])
447 # do the following for every entry that uses the controlled vocabulary!
448 # this should be improved!
449 if db_obj.country:
450 db_stationmeta.country = get_value_from_str(toardb.toardb.CN_vocabulary, db_obj.country)
451 if db_obj.timezone:
452 db_stationmeta.timezone = get_value_from_str(toardb.toardb.TZ_vocabulary, db_obj.timezone)
453 if db_obj.type:
454 db_stationmeta.type = get_value_from_str(toardb.toardb.ST_vocabulary, db_obj.type)
455 if db_obj.type_of_area:
456 db_stationmeta.type_of_area = get_value_from_str(toardb.toardb.TA_vocabulary, db_obj.type_of_area)
457 if db_obj.coordinate_validation_status:
458 db_stationmeta.coordinate_validation_status = get_value_from_str(toardb.toardb.CV_vocabulary, db_obj.coordinate_validation_status)
459 result = db.commit()
460 # store roles and update association table
461 if roles_data:
462 # prepare changelog entry/entries
463 if not no_log:
464 description = description + f"; add role"
465 db_old_roles = get_stationmeta_roles(db, station_id)
466 old_value = old_value + "'roles': {"
467 for oldr in db_old_roles:
468 old_role = get_stationmeta_role_by_id(db, oldr.role_id)
469 old_value=old_value + "{'role': '" + get_str_from_value(toardb.toardb.RC_vocabulary,old_role.role) + \
470 "', 'status': '" + get_str_from_value(toardb.toardb.RS_vocabulary,old_role.status) + \
471 "', 'contact_id': " + str(old_role.contact_id) + '}'
472 if (number_of_commas == 0):
473 old_value=old_value + '}'
474 else:
475 old_value=old_value + ','
476 number_of_commas = number_of_commas - 1
477 for r in roles_data:
478 db_role = models.StationmetaRole(**r)
479 db_role.role = get_value_from_str(toardb.toardb.RC_vocabulary,db_role.role)
480 db_role.status = get_value_from_str(toardb.toardb.RS_vocabulary,db_role.status)
481 # check whether role is already present in database
482 db_object = get_unique_stationmeta_role(db, db_role.role, db_role.contact_id, db_role.status)
483 if db_object:
484 role_id = db_object.id
485 else:
486 db.add(db_role)
487 db.commit()
488 db.refresh(db_role)
489 role_id = db_role.id
490 db.execute(insert(stationmeta_core_stationmeta_roles_table).values(station_id=station_id, role_id=role_id))
491 db.commit()
492 # store annotations and update association table
493 if annotations_data:
494 # prepare changelog entry/entries
495 if not no_log:
496 description = description + f"; add annotation"
497 db_old_annotations = get_stationmeta_annotations(db, station_id)
498 for olda in db_old_annotations:
499 old_annotation = get_stationmeta_annotation_by_id(db, olda.annotation_id)
500 old_value=old_value + "{'kind': '" + get_str_from_value(toardb.toardb.RC_vocabulary,old_role.kind) + \
501 "', 'text': '" + get_str_from_value(toardb.toardb.RS_vocabulary,old_role.status) + \
502 "', 'date_added': '" + get_str_from_value(toardb.toardb.RS_vocabulary,old_role.status) + \
503 "', 'approved': '" + get_str_from_value(toardb.toardb.RS_vocabulary,old_role.status) + \
504 "', 'contributor_id': " + str(old_role.contact_id) + '}'
505 if (number_of_commas == 0):
506 old_value=old_value + '}'
507 else:
508 old_value=old_value + ','
509 number_of_commas = number_of_commas - 1
510 for a in annotations_data:
511 db_annotation = models.StationmetaAnnotation(**a)
512 db_annotation.kind = get_value_from_str(toardb.toardb.AK_vocabulary,db_annotation.kind)
513 # check whether annotation is already present in database
514 db_object = get_unique_stationmeta_annotation(db, db_annotation.text, db_annotation.contributor_id)
515 if db_object:
516 annotation_id = db_object.id
517 else:
518 db.add(db_annotation)
519 db.commit()
520 db.refresh(db_annotation)
521 annotation_id = db_annotation.id
522 db.execute(insert(stationmeta_core_stationmeta_annotations_table).values(station_id=station_id, annotation_id=annotation_id))
523 db.commit()
524 # patch aux_images
525 if aux_images_data:
526 for i in aux_images_data:
527 db_aux_image = models.StationmetaAuxImage(**i)
528 db_aux_image.station_id = station_id
529 db.add(db_aux_image)
530 db.commit()
531 db.refresh(db_aux_image)
532 # patch aux_docs
533 if aux_docs_data:
534 for d in aux_docs_data:
535 db_aux_doc = models.StationmetaAuxDoc(**d)
536 db_aux_doc.station_id = station_id
537 db.add(db_aux_doc)
538 db.commit()
539 db.refresh(db_aux_doc)
540 # patch aux_urls
541 if aux_urls_data:
542 for u in aux_urls_data:
543 db_aux_url = models.StationmetaAuxUrl(**u)
544 db_aux_url.station_id = station_id
545 db.add(db_aux_url)
546 db.commit()
547 db.refresh(db_aux_url)
548 # patch globalmeta
549 if globalmeta_data:
550 db_globalmeta = get_stationmeta_global(db,station_id)
551 old_global = {}
552 if db_globalmeta:
553 # combine old and new data:
554 for key, value in globalmeta_data.items():
555 # "if value" would also skip a value of 0.0!!!
556 if value != None:
557 old_value = getattr(db_globalmeta, key)
558 # do the following for every entry that uses the controlled vocabulary!
559 # this should be improved!
560 if key == "climatic_zone_year2016":
561 value = get_value_from_str(toardb.toardb.CZ_vocabulary, value)
562 old_value = get_str_from_value(toardb.toardb.CZ_vocabulary, old_value)
563 elif key == "toar1_category":
564 value = get_value_from_str(toardb.toardb.TC_vocabulary, value)
565 old_value = get_str_from_value(toardb.toardb.TC_vocabulary, old_value)
566 elif key == "toar2_category":
567 value = get_value_from_str(toardb.toardb.TA_vocabulary, value)
568 old_value = get_str_from_value(toardb.toardb.TA_vocabulary, old_value)
569 elif key == "htap_region_tier1_year2010":
570 value = get_value_from_str(toardb.toardb.TR_vocabulary, value)
571 old_value = get_str_from_value(toardb.toardb.TR_vocabulary, old_value)
572 elif key == "dominant_landcover_year2012":
573 value = get_value_from_str(toardb.toardb.LC_vocabulary, value)
574 old_value = get_str_from_value(toardb.toardb.LC_vocabulary, old_value)
575 elif key == "landcover_description_25km_year2012":
576 value = get_abbreviation_from_code_description(toardb.toardb.LC_vocabulary, value)
577 old_value = get_code_description_from_abbreviation(toardb.toardb.LC_vocabulary, old_value)
578 elif key == "dominant_ecoregion_year2017":
579 value = get_value_from_str(toardb.toardb.ER_vocabulary, value)
580 old_value = get_str_from_value(toardb.toardb.ER_vocabulary, old_value)
581 elif key == "ecoregion_description_25km_year2017":
582 value = get_abbreviation_from_code_description(toardb.toardb.ER_vocabulary, value)
583 old_value = get_code_description_from_abbreviation(toardb.toardb.ER_vocabulary, old_value)
584 else:
585 pass
586 old_global[key] = old_value
587 setattr(db_globalmeta, key, value)
588 else:
589 db_globalmeta = models.StationmetaGlobal(**globalmeta_data)
590 db.add(db_globalmeta)
591 db.commit()
592 stationmeta_dict2 = {k: v for k, v in stationmeta_dict2['globalmeta'].items() if v is not None}
593 old_value = str(old_global)
594 if not no_log:
595 db_changelog = StationmetaChangelog(description=description, station_id=station_id, author_id=author_id, type_of_change=type_of_change,
596 old_value=old_value, new_value=str(stationmeta_dict2))
597 db.add(db_changelog)
598 db.commit()
599 # there's a mismatch with coordinates --> how to automatically switch back and forth?!
600 db_stationmeta.coordinates = tmp_coordinates
601 status_code = 200
602 message={"message": f'patched stationmeta record for station_id {station_id}'}
603 message["station_id"] = station_id
604 return JSONResponse(status_code=status_code, content=message)
607def delete_stationmeta_field(db: Session, station_id: int, field: str, author_id: int):
608 # id can never be deleted (and of course also not changed)!!!
609 # there are mandatory fields (from stationmeta_core), that cannot be deleted!
610 # --> set these to their default value
611 new_value = ""
612 field_table = {'roles': stationmeta_core_stationmeta_roles_table,
613 'annotations': stationmeta_core_stationmeta_annotations_table}
614# 'aux_images': stationmeta_core_stationmeta_aux_images_table,
615# 'aux_docs': stationmeta_core_stationmeta_aux_docs_table,
616# 'aux_urls': stationmeta_core_stationmeta_aux_urls_table,
617 if (field in field_table):
618 db.execute(delete(field_table[field]).where(field_table[field].c.station_id==station_id))
619 new_value = f"'{field}': []"
620 # problem with automatic conversion of coordinates (although not explicitly fetched from database)
621 # ==> next two lines are a workaround
622 db_stationmeta = db.query(models.StationmetaCore).get(station_id)
623 tmp_coordinates = db_stationmeta.coordinates
624 db_stationmeta.coordinates = get_geom_from_coordinates(db_stationmeta.coordinates)
625 # also write changelog entry
626 type_of_change = get_value_from_str(toardb.toardb.CL_vocabulary,"SingleValue")
627 description=f"delete field {field}"
628 old_value = get_field_from_record(db, station_id, field, db_stationmeta)
629 db_changelog = StationmetaChangelog(description=description, station_id=station_id, author_id=author_id, type_of_change=type_of_change,
630 old_value=old_value, new_value=new_value)
631 db.add(db_changelog)
632 db.commit()
633 # there's a mismatch with coordinates --> how to automatically switch back and forth?!
634 db_stationmeta.coordinates = tmp_coordinates
635 # hotfix: db_stationmeta.global needs to be retranslated to the dict that is understood by StationmetaGlobal
636 db_stationmeta.globalmeta = None
637 return db_stationmeta