Coverage for toardb/stationmeta/stationmeta.py: 96%

1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH 

2# SPDX-License-Identifier: MIT 

3 

4""" 

5Simple test API for stationmeta management 

6""" 

7from typing import List 

8from fastapi import APIRouter, Depends, HTTPException, Body, Request 

9from sqlalchemy.orm import Session, sessionmaker 

10from sqlalchemy.engine import Engine 

11from . import crud, schemas 

12from toardb.utils.database import get_db, get_engine 

13from toardb.utils.utils import ( 

14 get_admin_access_rights, 

15 get_station_md_change_access_rights 

16) 

17 

18 

19router = APIRouter() 

20 

21# CRUD: create, retrieve, update, delete 

22 

23# 1. retrieve 

24 

25#get all entries of table stationmeta 

26@router.get('/stationmeta/', response_model=List[schemas.Stationmeta | schemas.StationmetaFields], response_model_exclude_none=True, response_model_exclude_unset=True) 

27def get_all_stationmeta(request: Request, db: Session = Depends(get_db)): 

28 return crud.get_all_stationmeta(db, path_params=request.path_params, query_params=request.query_params) 

29 

30 

31#get all core metadata of one station 

32@router.get('/stationmeta/{station_code}', response_model=schemas.Stationmeta | schemas.StationmetaFields, response_model_exclude_none=True, response_model_exclude_unset=True) 

33def get_stationmeta(station_code: str, fields: str = None, db: Session = Depends(get_db)): 

34 db_stationmeta = crud.get_stationmeta(db, station_code=station_code, fields=fields) 

35 if db_stationmeta is None: 

36 raise HTTPException(status_code=404, detail=f"Metadata for station '{station_code}' not found.") 

37 return db_stationmeta 

38 

39 

40#get all core metadata of one station (given its ID) 

41@router.get('/stationmeta/id/{station_id}', response_model=schemas.Stationmeta, response_model_exclude_none=True, response_model_exclude_unset=True) 

42def get_stationmeta_by_id(station_id: int, db: Session = Depends(get_db)): 

43 db_stationmeta = crud.get_stationmeta_by_id(db, station_id=station_id) 

44 if db_stationmeta is None: 

45 raise HTTPException(status_code=404, detail=f"Metadata for station with ID '{station_id}' not found.") 

46 return db_stationmeta 

47 

48 

49#to be deleted? 

50@router.get('/stationmeta_changelog/{station_id}', response_model=List[schemas.StationmetaChangelog]) 

51def get_stationmeta_changelog(station_id: int, db: Session = Depends(get_db)): 

52 db_changelog = crud.get_stationmeta_changelog(db, station_id=station_id) 

53 return db_changelog 

54 

55 

56# post and patch only via authorization by Helmholtz AAI 

57 

58# 2. create 

59 

60@router.post('/stationmeta/') 

61async def create_stationmeta_core(request: Request, 

62 stationmeta: schemas.StationmetaCreate = Body(..., embed = True), 

63 force: bool = False, 

64 access: dict = Depends(get_admin_access_rights), 

65 db: Session = Depends(get_db), 

66 engine: Engine = Depends(get_engine)): 

67 # check whether the post is authorized (401: Unauthorized) 

68 if access['status_code'] == 200: 

69 for station_code in stationmeta.codes: 

70 db_stationmeta_core= crud.get_stationmeta_core(db, station_code=station_code) 

71 if db_stationmeta_core: 

72 raise HTTPException(status_code=443, detail={"message":"Station already registered!", "station_id": db_stationmeta_core.id}) 

73 return crud.create_stationmeta(db=db, engine=engine, stationmeta=stationmeta, 

74 author_id=access['auth_user_id'], force=force) 

75 else: 

76 raise HTTPException(status_code=401, detail="Unauthorized.") 

77 

78 

79# 3. update 

80 

81#@router.patch('/stationmeta/{station_code}', response_model=schemas.StationmetaPatch) 

82@router.patch('/stationmeta/{station_code}') 

83def patch_stationmeta_core(request: Request, 

84 station_code: str, 

85 stationmeta: schemas.StationmetaPatch = Body(..., embed = True), 

86 access: dict = Depends(get_station_md_change_access_rights), 

87 db: Session = Depends(get_db)): 

88 # check whether the patch is authorized (401: Unauthorized) 

89 if access['status_code'] == 200: 

90 query_params=request.query_params 

91 # check, if description has been given in query_params! 

92 if not "description" in query_params: 

93 raise HTTPException(status_code=404, detail="description text ist missing.") 

94 else: 

95 description = query_params.get("description") 

96 db_stationmeta = crud.get_stationmeta(db, station_code=station_code) 

97 if db_stationmeta is None: 

98 raise HTTPException(status_code=404, detail="Station for patching not found.") 

99 return crud.patch_stationmeta(db=db, description=description, 

100 station_id=db_stationmeta.id, stationmeta=stationmeta, 

101 author_id=access['auth_user_id']) 

102 else: 

103 raise HTTPException(status_code=401, detail="Unauthorized.") 

104 

105 

106@router.patch('/stationmeta/delete_field/{station_code}', response_model=schemas.StationmetaBase, response_model_exclude_none=True, response_model_exclude_unset=True) 

107def delete_field_from_stationmeta_core(request: Request, 

108 station_code: str, 

109 field: str, 

110 access: dict = Depends(get_station_md_change_access_rights), 

111 db: Session = Depends(get_db)): 

112 # check whether the patch is authorized (401: Unauthorized) 

113 if access['status_code'] == 200: 

114 db_stationmeta = crud.get_stationmeta(db, station_code=station_code) 

115 if db_stationmeta is None: 

116 raise HTTPException(status_code=404, detail="Station for deleting field not found.") 

117 return crud.delete_stationmeta_field(db=db, station_id=db_stationmeta.id, field=field, 

118 author_id=access['auth_user_id']) 

119 else: 

120 raise HTTPException(status_code=401, detail="Unauthorized.") 

121