Coverage for toardb/variables/crud.py: 100%

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

2# SPDX-License-Identifier: MIT 

3 

4""" 

5Create, Read, Update, Delete functionality 

6 

7""" 

8 

9from fastapi.responses import JSONResponse 

10from sqlalchemy import text 

11from sqlalchemy.orm import Session 

12from . import models, schemas 

13from toardb.utils.utils import create_filter 

14 

15 

16def get_variable(db: Session, variable_id: int): 

17 return ( 

18 db.query(models.Variable) 

19 .filter(models.Variable.id == variable_id) 

20 .first() 

21 ) 

22 

23 

24def get_variable_by_name(db: Session, name: str): 

25 return ( 

26 db.query(models.Variable) 

27 .filter(models.Variable.name == name.lower()) 

28 .first() 

29 ) 

30 

31 

32def get_variables(db: Session, path_params, query_params): 

33 try: 

34 limit, offset, fields, format, filters = create_filter( 

35 query_params, "variables" 

36 ) 

37 v_filter = filters["v_filter"] 

38 except (KeyError, ValueError) as e: 

39 status_code = 400 

40 return JSONResponse(status_code=status_code, content=str(e)) 

41 

42 if fields: 

43 fields2 = fields.split(",") 

44 db_objects_l = ( 

45 db.query(*map(text, fields2)) 

46 .select_from(models.Variable) 

47 .filter(text(v_filter)) 

48 .order_by(models.Variable.id) 

49 .limit(limit) 

50 .offset(offset) 

51 ) 

52 db_objects = [] 

53 for db_object_immut in db_objects_l: 

54 db_object = dict(zip(fields.split(","), db_object_immut)) 

55 db_objects.append(db_object) 

56 else: 

57 db_objects = ( 

58 db.query(models.Variable) 

59 .filter(text(v_filter)) 

60 .order_by(models.Variable.id) 

61 .offset(offset) 

62 .limit(limit) 

63 .all() 

64 ) 

65 return db_objects 

66 

67 

68def create_variable(db: Session, variable: schemas.VariableCreate): 

69 db_variable = models.Variable(**variable.dict()) 

70 db.add(db_variable) 

71 db.commit() 

72 db.refresh(db_variable) 

73 return db_variable