""" Simple test API for variable management """ from typing import List from fastapi import FastAPI, Depends, HTTPException, APIRouter from sqlalchemy.orm import Session from toardb.utils.database import ToarDbSession, engine from toardb.variables import variables from toardb.contacts import contacts from toardb.stationmeta import stationmeta from toardb.timeseries import timeseries from toardb.data import data from collections import namedtuple from pydantic import BaseSettings app = FastAPI() @app.get("/controlled_vocabulary/") async def info(): controlled_vocabulary = { "Role Codes": settings.RC_vocab, "Role Status": settings.RS_vocab, "Kind of Organizations": settings.OK_vocab, "Data Access Rights": settings.DA_vocab, "Sampling Frequencies": settings.SF_vocab, "Aggregation Types": settings.AT_vocab, "Data Sources": settings.DS_vocab, "Measurement Methods": settings.MM_vocab, "Climatic Zones": settings.CZ_vocab, "Coordinate Validity": settings.CV_vocab, "Station Types": settings.ST_vocab, "Station Type of Area": settings.TA_vocab, "Station TOAR Categories": settings.TC_vocab, "Station HTAP Regions": settings.TR_vocab, "Station Dominant Landcover Types": settings.DL_vocab, "Result Types": settings.RT_vocab, "Data Flags": settings.DF_vocab, } return controlled_vocabulary @app.get("/controlled_vocabulary/{name}") async def info(name: str): controlled_vocabulary = { "Role Codes": settings.RC_vocab, "Role Status": settings.RS_vocab, "Kind of Organizations": settings.OK_vocab, "Data Access Rights": settings.DA_vocab, "Sampling Frequencies": settings.SF_vocab, "Aggregation Types": settings.AT_vocab, "Data Sources": settings.DS_vocab, "Measurement Methods": settings.MM_vocab, "Climatic Zones": settings.CZ_vocab, "Coordinate Validity": settings.CV_vocab, "Station Types": settings.ST_vocab, "Station Type of Area": settings.TA_vocab, "Station TOAR Categories": settings.TC_vocab, "Station HTAP Regions": settings.TR_vocab, "Station Dominant Landcover Types": settings.DL_vocab, "Result Types": settings.RT_vocab, "Data Flags": settings.DF_vocab, } return controlled_vocabulary[name] db_stats= { "users": 358, "stations": 13250, "time-series": 103678, "data records": 12010313565, } @app.get("/database_statistics/") async def info(): return db_stats @app.get('/database_statistics/{name}') async def info(name: str): return db_stats[name] # Dependency def get_db(): try: db = ToarDbSession() yield db finally: db.close() app.include_router(variables.router) app.include_router(contacts.router) app.include_router(stationmeta.router) app.include_router(timeseries.router) app.include_router(data.router) # get the controlled vocabulary from table def __get_enum_dict(fake_cur, table_name): fake_cur.execute("select * from "+table_name) res = fake_cur.fetchall() Enumdict=namedtuple("Dict",["value","string","display_str"]) enum_dict = [] for entry in res: enum_dict.append(Enumdict(*entry)) return enum_dict @app.on_event("startup") # will be executed before application *starts* # ==> again: at this point no database connection available! # (and also all tables are unknown) # how can I do pytests with this?!?! # ==> https://fastapi.tiangolo.com/advanced/testing-events/ # I did not try yet, if test database is created before # startup_event or after (for the letter, the hint in the link # will not work) async def startup_event(): fake_conn = engine.raw_connection() fake_cur = fake_conn.cursor() RC_vocabulary = __get_enum_dict(fake_cur, "rc_vocabulary") RS_vocabulary = __get_enum_dict(fake_cur, "rs_vocabulary") OK_vocabulary = __get_enum_dict(fake_cur, "ok_vocabulary") DA_vocabulary = __get_enum_dict(fake_cur, "da_vocabulary") SF_vocabulary = __get_enum_dict(fake_cur, "sf_vocabulary") AT_vocabulary = __get_enum_dict(fake_cur, "at_vocabulary") DS_vocabulary = __get_enum_dict(fake_cur, "ds_vocabulary") MM_vocabulary = __get_enum_dict(fake_cur, "mm_vocabulary") CZ_vocabulary = __get_enum_dict(fake_cur, "cz_vocabulary") CV_vocabulary = __get_enum_dict(fake_cur, "cv_vocabulary") ST_vocabulary = __get_enum_dict(fake_cur, "st_vocabulary") TA_vocabulary = __get_enum_dict(fake_cur, "ta_vocabulary") TC_vocabulary = __get_enum_dict(fake_cur, "tc_vocabulary") TR_vocabulary = __get_enum_dict(fake_cur, "tr_vocabulary") DL_vocabulary = __get_enum_dict(fake_cur, "dl_vocabulary") RT_vocabulary = __get_enum_dict(fake_cur, "rt_vocabulary") DF_vocabulary = __get_enum_dict(fake_cur, "df_vocabulary") Enumdict=namedtuple("Dict",["value","string","display_str"]) fake_conn = engine.raw_connection() fake_cur = fake_conn.cursor() RC_vocabulary = __get_enum_dict(fake_cur, "rc_vocabulary") RS_vocabulary = __get_enum_dict(fake_cur, "rs_vocabulary") OK_vocabulary = __get_enum_dict(fake_cur, "ok_vocabulary") DA_vocabulary = __get_enum_dict(fake_cur, "da_vocabulary") SF_vocabulary = __get_enum_dict(fake_cur, "sf_vocabulary") AT_vocabulary = __get_enum_dict(fake_cur, "at_vocabulary") DS_vocabulary = __get_enum_dict(fake_cur, "ds_vocabulary") MM_vocabulary = __get_enum_dict(fake_cur, "mm_vocabulary") CZ_vocabulary = __get_enum_dict(fake_cur, "cz_vocabulary") CV_vocabulary = __get_enum_dict(fake_cur, "cv_vocabulary") ST_vocabulary = __get_enum_dict(fake_cur, "st_vocabulary") TA_vocabulary = __get_enum_dict(fake_cur, "ta_vocabulary") TC_vocabulary = __get_enum_dict(fake_cur, "tc_vocabulary") TR_vocabulary = __get_enum_dict(fake_cur, "tr_vocabulary") DL_vocabulary = __get_enum_dict(fake_cur, "dl_vocabulary") RT_vocabulary = __get_enum_dict(fake_cur, "rt_vocabulary") DF_vocabulary = __get_enum_dict(fake_cur, "df_vocabulary") class Settings(BaseSettings): Enumdict=namedtuple("Dict",["value","string","display_str"]) RC_vocab = RC_vocabulary RS_vocab = RS_vocabulary OK_vocab = OK_vocabulary DA_vocab = DA_vocabulary SF_vocab = SF_vocabulary AT_vocab = AT_vocabulary DS_vocab = DS_vocabulary MM_vocab = MM_vocabulary CZ_vocab = CZ_vocabulary CV_vocab = CV_vocabulary ST_vocab = ST_vocabulary TA_vocab = TA_vocabulary TC_vocab = TC_vocabulary TR_vocab = TR_vocabulary DL_vocab = DL_vocabulary RT_vocab = RT_vocabulary DF_vocab = DF_vocabulary settings = Settings()