""" Simple test API for stationmeta management """ from typing import List from fastapi import APIRouter, Depends, HTTPException, Body from sqlalchemy.orm import Session, sessionmaker from . import crud, schemas from sqlalchemy import create_engine from toardb.utils.database import DATABASE_URL, get_db router = APIRouter() # plain views to post and get stationmeta #get all entries of table stationmeta @router.get('/stationmeta_core/', response_model=List[schemas.StationmetaCore]) def get_all_stationmeta_core(skip: int = 0, limit: int = None, db: Session = Depends(get_db)): stationmeta_core = crud.get_all_stationmeta_core(db, skip=skip, limit=limit) return stationmeta_core # the same as above, but nested view #get all entries of table stationmeta @router.get('/stationmeta/', response_model=List[schemas.Stationmeta]) def get_all_stationmeta(skip: int = 0, limit: int = None, db: Session = Depends(get_db)): stationmeta = crud.get_all_stationmeta(db, skip=skip, limit=limit) return stationmeta #get all core metadata of one station @router.get('/stationmeta_core/{station_code}', response_model=schemas.StationmetaCore) def get_stationmeta_core(station_code: str, db: Session = Depends(get_db)): db_stationmeta_core = crud.get_stationmeta_core(db, station_code=station_code) if db_stationmeta_core is None: raise HTTPException(status_code=404, detail="Data not found.") return db_stationmeta_core #some more gets to be tested: # - get stationmeta_global # - get stationmeta_aux # - ... @router.post('/stationmeta_core/', response_model=schemas.StationmetaCore) #curl -X POST -H "Content-Type:application/json" -d '{"stationmeta_core": {"codes":["ttt3","ttt4"],"name":"Test_China","coordinates":{"lat":36.256,"lng":17.106,"alt":1534.0},"country":"China","state":"Shandong Sheng","coordinate_validation_status":0,"coordinate_validation_date":"2020-03-11T12:22:18.047974+01:00","type_of_environment":0,"type_of_area":0,"category":"","timezone":"", "coordinate_validator_id": 1, "additional_metadata":"{}"}}' "http://127.0.0.1:8000/stationmeta_core/" # The following command was not working as long as the upload via Body was defined. # See bug report: https://github.com/tiangolo/fastapi/issues/300 # (Although this seems to be fixed in the meantime, it is not working in my FastAPI version.) #def create_stationmeta_core(stationmeta_core: schemas.StationmetaCoreCreate, db: Session = Depends(get_db)): def create_stationmeta_core(stationmeta_core: schemas.StationmetaCoreCreate = Body(..., embed = True), db: Session = Depends(get_db)): # for the moment, just check the first code of station's codes db_stationmeta_core= crud.get_stationmeta_core(db, station_code=stationmeta_core.codes[0]) if db_stationmeta_core: raise HTTPException(status_code=400, detail="Station already registered.") return crud.create_stationmeta_core(db=db, stationmeta_core=stationmeta_core)