Coverage for yaptide/routes/user_routes.py: 91%

1import logging 

2from enum import Enum 

3 

4from flask import request 

5from flask_restful import Resource 

6from marshmallow import Schema, fields 

7from sqlalchemy import asc, desc 

8 

9from yaptide.persistence.models import SimulationModel, UserModel 

10from yaptide.routes.utils.decorators import requires_auth 

11from yaptide.routes.utils.response_templates import (error_validation_response, 

12 yaptide_response) 

13 

14DEFAULT_PAGE_SIZE = 6 # default number of simulations per page 

15DEFAULT_PAGE_IDX = 1 # default page index 

16 

17 

18class OrderType(Enum): 

19 """Order type""" 

20 

21 ASCEND = "ascend" 

22 DESCEND = "descend" 

23 

24 

25class OrderBy(Enum): 

26 """Order by column""" 

27 

28 START_TIME = "start_time" 

29 END_TIME = "end_time" 

30 

31 

32class UserSimulations(Resource): 

33 """Class responsible for returning user's simulations' basic infos""" 

34 

35 class APIParametersSchema(Schema): 

36 """Class specifies API parameters""" 

37 

38 page_size = fields.Integer(load_default=DEFAULT_PAGE_SIZE) 

39 page_idx = fields.Integer(load_default=DEFAULT_PAGE_IDX) 

40 order_by = fields.String(load_default=OrderBy.START_TIME.value) 

41 order_type = fields.String(load_default=OrderType.DESCEND.value) 

42 

43 @staticmethod 

44 @requires_auth() 

45 def get(user: UserModel): 

46 """Method returning simulations from the database""" 

47 schema = UserSimulations.APIParametersSchema() 

48 params_dict: dict = schema.load(request.args) 

49 logging.info('User %s requested simulations with parameters: %s', user.username, params_dict) 

50 

51 # Query the database for the paginated results 

52 sorting = desc if params_dict['order_type'] == OrderType.DESCEND.value else asc 

53 query = SimulationModel.query.\ 

54 filter(SimulationModel.job_id != None).\ 

55 filter_by(user_id=user.id).\ 

56 order_by(sorting(params_dict['order_by'])) 

57 pagination = query.paginate(page=params_dict['page_idx'], per_page=params_dict['page_size'], error_out=False) 

58 simulations = pagination.items 

59 

60 result = { 

61 'simulations': [ 

62 { 

63 'title': simulation.title, 

64 'job_id': simulation.job_id, 

65 'start_time': simulation.start_time, 

66 # submission time, when user send the request to the backend - jobs may start much later than that 

67 'end_time': simulation.end_time, 

68 # end time, when the all jobs are finished and results are merged 

69 'metadata': { 

70 'platform': simulation.platform, 

71 'server': 'Yaptide', 

72 'input_type': simulation.input_type, 

73 'sim_type': simulation.sim_type 

74 } 

75 } 

76 for simulation in simulations], 

77 'page_count': pagination.pages, 

78 'simulations_count': pagination.total, 

79 } 

80 return yaptide_response(message='User Simulations', code=200, content=result) 

81 

82 

83class UserUpdate(Resource): 

84 """Class responsible for updating the user""" 

85 

86 @staticmethod 

87 @requires_auth() 

88 def post(user: UserModel): 

89 """Updates user with provided parameters""" 

90 json_data: dict = request.get_json(force=True) 

91 if not json_data: 

92 return error_validation_response() 

93 return yaptide_response(message=f'User {user.username} updated', code=202)