Coverage for toardb / toardb.py: 88%
180 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"""
5API for TOAR-II database
6"""
8from collections import namedtuple
9from fastapi.responses import Response
10import io
11import functools
12import json
13import pandas as pd
15from pydantic import BaseSettings
16from typing import List
17from fastapi import FastAPI, Depends, HTTPException, APIRouter, \
18 Request
19from sqlalchemy import text
20from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
21from fastapi.staticfiles import StaticFiles
22from sqlalchemy.orm import Session
23from starlette.responses import FileResponse
24from fastapi.templating import Jinja2Templates
26from pyinstrument import Profiler
27from pyinstrument.renderers.html import HTMLRenderer
28from pyinstrument.renderers.speedscope import SpeedscopeRenderer
29from pathlib import Path
30import time
32from toardb.utils.database import ToarDbSession, engine, get_db
33from toardb.utils.settings import base_url
34from toardb.utils.utils import normalize_metadata
35from sqlalchemy import func
36from toardb.auth_user.models import AuthUser
37from toardb.timeseries.models_core import Timeseries
39from sqlalchemy import text
40from toardb.auth_user import auth_user
41from toardb.variables import variables
42from toardb.contacts import contacts
43from toardb.stationmeta import stationmeta, models
44from toardb.timeseries import timeseries
45from toardb.data import data
47RC_vocabulary = 0
48RS_vocabulary = 0
49AK_vocabulary = 0
50OK_vocabulary = 0
51SF_vocabulary = 0
52AT_vocabulary = 0
53OT_vocabulary = 0
54DO_vocabulary = 0
55CZ_vocabulary = 0
56CV_vocabulary = 0
57CN_vocabulary = 0
58TZ_vocabulary = 0
59ST_vocabulary = 0
60TA_vocabulary = 0
61TC_vocabulary = 0
62TR_vocabulary = 0
63LC_vocabulary = 0
64ER_vocabulary = 0
65RT_vocabulary = 0
66DF_vocabulary = 0
67CL_vocabulary = 0
68CS_vocabulary = 0
69KS_vocabulary = 0
70CT_vocabulary = 0
71controlled_fields = 0
73app = FastAPI()
76app.mount("/static", StaticFiles(directory="static"), name="_static")
77templates = Jinja2Templates(directory="templates")
80# check more on https://pyinstrument.readthedocs.io/en/latest/guide.html#profile-a-web-request-in-fastapi
81@app.middleware("http")
82async def profile_request(request: Request, call_next):
83 profile_type_to_ext = {"html": "html", "json": "json"}
84 profile_type_to_renderer = {
85 "html": HTMLRenderer,
86 "json": SpeedscopeRenderer,
87 }
88 if request.query_params.get("profile", False): # pragma: no cover
89 current_dir = Path(__file__).parent
90 profile_type = request.query_params.get("profile_format", "html")
91 with Profiler(interval=0.001, async_mode="enabled") as profiler:
92 response = await call_next(request)
93 ext = profile_type_to_ext[profile_type]
94 renderer = profile_type_to_renderer[profile_type]()
95 with open(current_dir / f"../profile.{ext}", "a") as outfile:
96 outfile.write(profiler.output(renderer=renderer))
97 return response
98 return await call_next(request)
101# check more on https://fastapi.tiangolo.com/tutorial/middleware/
102@app.middleware("http")
103async def add_process_time_header(request: Request, call_next):
104 if request.query_params.get("timing", False): # pragma: no cover
105 current_dir = Path(__file__).parent
106 start_time = time.time()
107 response = await call_next(request)
108 with open(current_dir / f"../timing.txt", "a") as outfile:
109 outfile.write("{} s: {}\n".format(time.time() - start_time, request.url))
110 return response
111 return await call_next(request)
114@app.middleware("http")
115async def response_to_csv(request: Request, call_next):
116 response = await call_next(request)
117 if ((response.status_code != 200) or
118 (request['path'].startswith(('/data/timeseries_merged/','/data/timeseries/','/data/timeseries_with_staging/','/ontology/','/controlled_vocabulary/','/auth_user'))) or
119 (request.query_params.get('format') != 'csv')):
120 return response
121 # from: https://stackoverflow.com/a/71883126
122 response_body = b""
123 async for chunk in response.body_iterator:
124 response_body += chunk
125 list_response = json.loads(response_body)
126 if isinstance(list_response,dict):
127 list_response = [ list_response ]
128 metadata = pd.DataFrame([ normalize_metadata(resp_object) for resp_object in list_response ])
129 return Response(content=metadata.to_csv(index=False), status_code=response.status_code, media_type="text/csv")
131# add endpoints
132# additional yaml version of openapi.json
133@app.get('/openapi.yaml', include_in_schema=False)
134@functools.lru_cache()
135def read_openapi_yaml() -> Response:
136 openapi_json= app.openapi()
137 yaml_s = io.StringIO()
138 yaml.dump(openapi_json, yaml_s)
139 return Response(yaml_s.getvalue(), media_type='text/yaml')
142@app.get('/', response_class=HTMLResponse)
143def show_api(request: Request):
144 message = None
145 return templates.TemplateResponse("TOARDB_FASTAPI_Rest.html", {"request": request,
146 'message': message})
149@app.get("/ontology")
150def read_onto(format: str = 'xml'):
151 if format == 'owldoc':
152 return RedirectResponse(base_url + "/documentation/ontologies/v1.0/index.html")
153 else:
154 return FileResponse('static/ontology.xml')
157@app.get("/controlled_vocabulary/")
158def info():
159 controlled_vocabulary = {
160 "Role Code": RC_vocabulary,
161 "Role Status": RS_vocabulary,
162 "Kind Of Annotation": AK_vocabulary,
163 "Kind Of Organization": OK_vocabulary,
164 "Sampling Frequency": SF_vocabulary,
165 "Aggregation Type": AT_vocabulary,
166 "Data Origin Type": OT_vocabulary,
167 "Data Origin": DO_vocabulary,
168 "Climatic Zone 2019": CZ_vocabulary,
169 "Country Code": CN_vocabulary,
170 "Timezone": TZ_vocabulary,
171 "Station Coordinate Validity": CV_vocabulary,
172 "Station Type": ST_vocabulary,
173 "Station Type Of Area": TA_vocabulary,
174 "Station TOAR Category": TC_vocabulary,
175 "Station HTAP Region": TR_vocabulary,
176 "Station Landcover Type": LC_vocabulary,
177 "Station ECO Region Type": ER_vocabulary,
178 "Result Type": RT_vocabulary,
179 "Data Flag": DF_vocabulary,
180 "Type Of Change": CL_vocabulary,
181 "Absorption Cross Section": CS_vocabulary,
182 "Sampling Type": KS_vocabulary,
183 "Calibration Type": CT_vocabulary,
184 }
185 return controlled_vocabulary
187@app.get("/controlled_vocabulary/{name}")
188def info(name: str):
189 controlled_vocabulary = {
190 "role code": RC_vocabulary,
191 "role status": RS_vocabulary,
192 "kind of annotation": AK_vocabulary,
193 "kind of organization": OK_vocabulary,
194 "sampling frequency": SF_vocabulary,
195 "aggregation type": AT_vocabulary,
196 "data origin": DO_vocabulary,
197 "data origin type": OT_vocabulary,
198 "climatic zone 2019": CZ_vocabulary,
199 "coordinate validity": CV_vocabulary,
200 "country code": CN_vocabulary,
201 "timezone": TZ_vocabulary,
202 "station coordinate validity": CV_vocabulary,
203 "station type": ST_vocabulary,
204 "station type of area": TA_vocabulary,
205 "station toar category": TC_vocabulary,
206 "station htap region": TR_vocabulary,
207 "station landcover type": LC_vocabulary,
208 "station eco region type": ER_vocabulary,
209 "result type": RT_vocabulary,
210 "data flag": DF_vocabulary,
211 "type of change": CL_vocabulary,
212 "absorption cross section": CS_vocabulary,
213 "sampling type": KS_vocabulary,
214 "calibration type": CT_vocabulary,
215 }
216 if name.lower() in controlled_vocabulary.keys():
217 return controlled_vocabulary[name.lower()]
218 else:
219 status_code=200
220 message = f"No controlled vocabulary found for '{name}'"
221 return JSONResponse(status_code=status_code, content=message)
223def _db_statistics(db: Session):
224 # Compute basic database statistics
225 # stations: total stations in stationmeta_core
226 stations = db.query(func.count(models.StationmetaCore.id)).scalar() or 0
227 # time-series: total timeseries entries
228 time_series = db.query(func.count(Timeseries.id)).scalar() or 0
229 # users: total auth users
230 users = db.query(func.count(AuthUser.id)).scalar() or 0
231 # data records: since table public.data is tremendously huge, the total rows
232 # cannot be counted in a reasonable amout of time (like above);
233 # therefore, the estimated rumber of rows (reltuples) from pg_class
234 # is returned
235# data_records = (db.execute(
236# (text(
237# "SELECT reltuples::BIGINT FROM pg_class WHERE relname = 'data'")))
238# .scalar() or 0)
239 result = db.execute(text(
240 "SELECT reltuples::BIGINT FROM pg_class WHERE relname = 'data'"))
241 data_records = result.scalar() or 0
242 return {
243 "data records": data_records,
244 "stations": stations,
245 "time-series": time_series,
246 "users": users,
247 }
249@app.get("/database_statistics/")
250def stats_info(db: Session = Depends(get_db)):
251 return _db_statistics(db)
253@app.get('/database_statistics/{name}')
254def stats_info(name: str, db: Session = Depends(get_db)):
255 stats = _db_statistics(db)
256 return stats[name]
258@app.get("/geopeas_urls/")
259def geo_info(db: Session = Depends(get_db)):
260 res = db.query(models.StationmetaGlobalService).all()
261 return res
264# Dependency
265def get_db():
266 try:
267 db = ToarDbSession()
268 yield db
269 finally:
270 db.close()
272app.include_router(auth_user.router)
273app.include_router(variables.router)
274app.include_router(contacts.router)
275app.include_router(stationmeta.router)
276app.include_router(timeseries.router)
277app.include_router(data.router)
279# get the controlled vocabulary from table
280def __get_enum_dict(connection, table_name):
281 res = connection.execute("select * from "+table_name+" order by enum_val")
282 Enumdict=namedtuple("Dict",["value","string","display_str"])
283 enum_dict = []
284 for entry in res:
285 enum_dict.append(Enumdict(*entry))
286 return enum_dict
288@app.on_event("startup")
289# will be executed before application *starts*
290# ==> again: at this point no database connection available!
291# (and also all tables are unknown)
292def startup_event():
294 global RC_vocabulary
295 global RS_vocabulary
296 global AK_vocabulary
297 global OK_vocabulary
298 global SF_vocabulary
299 global AT_vocabulary
300 global OT_vocabulary
301 global DO_vocabulary
302 global CZ_vocabulary
303 global CV_vocabulary
304 global CN_vocabulary
305 global TZ_vocabulary
306 global ST_vocabulary
307 global TA_vocabulary
308 global TC_vocabulary
309 global TR_vocabulary
310 global LC_vocabulary
311 global ER_vocabulary
312 global RT_vocabulary
313 global DF_vocabulary
314 global CL_vocabulary
315 global CS_vocabulary
316 global KS_vocabulary
317 global CT_vocabulary
318 global controlled_fields
320 with engine.begin() as connection:
321 RC_vocabulary = __get_enum_dict(connection, "rc_vocabulary")
322 RS_vocabulary = __get_enum_dict(connection, "rs_vocabulary")
323 AK_vocabulary = __get_enum_dict(connection, "ak_vocabulary")
324 OK_vocabulary = __get_enum_dict(connection, "ok_vocabulary")
325 SF_vocabulary = __get_enum_dict(connection, "sf_vocabulary")
326 AT_vocabulary = __get_enum_dict(connection, "at_vocabulary")
327 OT_vocabulary = __get_enum_dict(connection, "ot_vocabulary")
328 DO_vocabulary = __get_enum_dict(connection, "do_vocabulary")
329 CZ_vocabulary = __get_enum_dict(connection, "cz_vocabulary")
330 CV_vocabulary = __get_enum_dict(connection, "cv_vocabulary")
331 CN_vocabulary = __get_enum_dict(connection, "cn_vocabulary")
332 TZ_vocabulary = __get_enum_dict(connection, "tz_vocabulary")
333 ST_vocabulary = __get_enum_dict(connection, "st_vocabulary")
334 TA_vocabulary = __get_enum_dict(connection, "ta_vocabulary")
335 TC_vocabulary = __get_enum_dict(connection, "tc_vocabulary")
336 TR_vocabulary = __get_enum_dict(connection, "tr_vocabulary")
337 LC_vocabulary = __get_enum_dict(connection, "lc_vocabulary")
338 ER_vocabulary = __get_enum_dict(connection, "er_vocabulary")
339 RT_vocabulary = __get_enum_dict(connection, "rt_vocabulary")
340 DF_vocabulary = __get_enum_dict(connection, "df_vocabulary")
341 CL_vocabulary = __get_enum_dict(connection, "cl_vocabulary")
342 CS_vocabulary = __get_enum_dict(connection, "cs_vocabulary")
343 KS_vocabulary = __get_enum_dict(connection, "ks_vocabulary")
344 CT_vocabulary = __get_enum_dict(connection, "ct_vocabulary")
346 # also get information, which database field is using which controlled vocabulary
348 controlled_fields = {}
349 res = connection.execute("SELECT c.conname, c1.conname FROM pg_catalog.pg_constraint c " +
350 "INNER JOIN pg_catalog.pg_constraint c1 ON c.confrelid=c1.conrelid " +
351 "AND c1.conname LIKE '%%enum_val%%' AND c.conname LIKE '%%_fk_%%_voc%%'")
352 for line in res:
353 field, voc = (line[0].split('_fk_'))
354 voc = voc.split('_')[0].upper()
355 controlled_fields[field] = f"{voc}_vocabulary"
356 engine.dispose()