Skip to content

user_routes

routes.user_routes

DEFAULT_PAGE_IDX module-attribute 

DEFAULT_PAGE_IDX = 1

DEFAULT_PAGE_SIZE module-attribute 

DEFAULT_PAGE_SIZE = 6

OrderBy

Bases: Enum

Order by column

Source code in yaptide/routes/user_routes.py 
26
27
28
29
30
class OrderBy(Enum):
    """Order by column"""

    START_TIME = "start_time"
    END_TIME = "end_time"

END_TIME class-attribute instance-attribute 

END_TIME = 'end_time'

START_TIME class-attribute instance-attribute 

START_TIME = 'start_time'

OrderType

Bases: Enum

Order type

Source code in yaptide/routes/user_routes.py 
19
20
21
22
23
class OrderType(Enum):
    """Order type"""

    ASCEND = "ascend"
    DESCEND = "descend"

ASCEND class-attribute instance-attribute 

ASCEND = 'ascend'

DESCEND class-attribute instance-attribute 

DESCEND = 'descend'

UserSimulations

Bases: Resource

Class responsible for returning user's simulations' basic infos

Source code in yaptide/routes/user_routes.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class UserSimulations(Resource):
    """Class responsible for returning user's simulations' basic infos"""

    class GetAPIParametersSchema(Schema):
        """Class specifies Get API parameters"""

        page_size = fields.Integer(load_default=DEFAULT_PAGE_SIZE)
        page_idx = fields.Integer(load_default=DEFAULT_PAGE_IDX)
        order_by = fields.String(load_default=OrderBy.START_TIME.value)
        order_type = fields.String(load_default=OrderType.DESCEND.value)

    class DeleteAPIParametersSchema(Schema):
        """Schema for DELETE method parameters"""

        job_id = fields.String(required=True)  # job_id is mandatory for DELETE

    @staticmethod
    @requires_auth()
    def get(user: UserModel):
        """Method returning simulations from the database"""
        schema = UserSimulations.GetAPIParametersSchema()
        params_dict: dict = schema.load(request.args)
        logging.info('User %s requested simulations with parameters: %s', user.username, params_dict)

        # Query the database for the paginated results
        sorting = desc if params_dict['order_type'] == OrderType.DESCEND.value else asc
        query = SimulationModel.query.\
            filter(SimulationModel.job_id != None).\
            filter_by(user_id=user.id).\
            order_by(sorting(params_dict['order_by']))
        pagination = query.paginate(page=params_dict['page_idx'], per_page=params_dict['page_size'], error_out=False)
        simulations = pagination.items

        result = {
            'simulations': [
                {
                    'title': simulation.title,
                    'job_id': simulation.job_id,
                    'start_time': simulation.start_time,
                    # submission time, when user send the request to the backend - jobs may start much later than that
                    'end_time': simulation.end_time,
                    # end time, when the all jobs are finished and results are merged
                    'metadata': {
                        'platform': simulation.platform,
                        'server': 'Yaptide',
                        'input_type': simulation.input_type,
                        'sim_type': simulation.sim_type
                    }
                } for simulation in simulations
            ],
            'page_count':
            pagination.pages,
            'simulations_count':
            pagination.total,
        }
        return yaptide_response(message='User Simulations', code=200, content=result)

    @staticmethod
    @requires_auth()
    def delete(user: UserModel):
        """Method deleting simulation from database"""
        schema = UserSimulations.DeleteAPIParametersSchema()
        errors: dict[str, list[str]] = schema.validate(request.args)
        if errors:
            return error_validation_response(content=errors)
        params_dict: dict = schema.load(request.args)

        job_id = params_dict['job_id']
        simulation = fetch_simulation_by_job_id(job_id)

        if simulation is None:
            return yaptide_response(message=f'Simulation with job_id={job_id} do not exist', code=404)

        if simulation.user_id != user.id:
            return yaptide_response(message='Unauthorized: You do not have permission to delete this simulation',
                                    code=401)

        # Simulation has to be completed/cancelled before deleting it.
        if simulation.job_state in (EntityState.UNKNOWN.value, EntityState.PENDING.value, EntityState.RUNNING.value):
            return yaptide_response(message=f'''Simulation with job_id={job_id} is currently running.
                  Please cancel simulation or wait for it to finish''',
                                    code=403)

        delete_object_from_db(simulation)
        return yaptide_response(message=f'Simulation with job_id={job_id} successfully deleted from database', code=200)

DeleteAPIParametersSchema

Bases: Schema

Schema for DELETE method parameters

Source code in yaptide/routes/user_routes.py 
44
45
46
47
class DeleteAPIParametersSchema(Schema):
    """Schema for DELETE method parameters"""

    job_id = fields.String(required=True)  # job_id is mandatory for DELETE
job_id class-attribute instance-attribute 
job_id = String(required=True)

GetAPIParametersSchema

Bases: Schema

Class specifies Get API parameters

Source code in yaptide/routes/user_routes.py 
36
37
38
39
40
41
42
class GetAPIParametersSchema(Schema):
    """Class specifies Get API parameters"""

    page_size = fields.Integer(load_default=DEFAULT_PAGE_SIZE)
    page_idx = fields.Integer(load_default=DEFAULT_PAGE_IDX)
    order_by = fields.String(load_default=OrderBy.START_TIME.value)
    order_type = fields.String(load_default=OrderType.DESCEND.value)
order_by class-attribute instance-attribute 
order_by = String(load_default=START_TIME.value)
order_type class-attribute instance-attribute 
order_type = String(load_default=DESCEND.value)
page_idx class-attribute instance-attribute 
page_idx = Integer(load_default=DEFAULT_PAGE_IDX)
page_size class-attribute instance-attribute 
page_size = Integer(load_default=DEFAULT_PAGE_SIZE)

delete staticmethod 

delete(user)

Method deleting simulation from database

Source code in yaptide/routes/user_routes.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@staticmethod
@requires_auth()
def delete(user: UserModel):
    """Method deleting simulation from database"""
    schema = UserSimulations.DeleteAPIParametersSchema()
    errors: dict[str, list[str]] = schema.validate(request.args)
    if errors:
        return error_validation_response(content=errors)
    params_dict: dict = schema.load(request.args)

    job_id = params_dict['job_id']
    simulation = fetch_simulation_by_job_id(job_id)

    if simulation is None:
        return yaptide_response(message=f'Simulation with job_id={job_id} do not exist', code=404)

    if simulation.user_id != user.id:
        return yaptide_response(message='Unauthorized: You do not have permission to delete this simulation',
                                code=401)

    # Simulation has to be completed/cancelled before deleting it.
    if simulation.job_state in (EntityState.UNKNOWN.value, EntityState.PENDING.value, EntityState.RUNNING.value):
        return yaptide_response(message=f'''Simulation with job_id={job_id} is currently running.
              Please cancel simulation or wait for it to finish''',
                                code=403)

    delete_object_from_db(simulation)
    return yaptide_response(message=f'Simulation with job_id={job_id} successfully deleted from database', code=200)

get staticmethod 

get(user)

Method returning simulations from the database

Source code in yaptide/routes/user_routes.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@staticmethod
@requires_auth()
def get(user: UserModel):
    """Method returning simulations from the database"""
    schema = UserSimulations.GetAPIParametersSchema()
    params_dict: dict = schema.load(request.args)
    logging.info('User %s requested simulations with parameters: %s', user.username, params_dict)

    # Query the database for the paginated results
    sorting = desc if params_dict['order_type'] == OrderType.DESCEND.value else asc
    query = SimulationModel.query.\
        filter(SimulationModel.job_id != None).\
        filter_by(user_id=user.id).\
        order_by(sorting(params_dict['order_by']))
    pagination = query.paginate(page=params_dict['page_idx'], per_page=params_dict['page_size'], error_out=False)
    simulations = pagination.items

    result = {
        'simulations': [
            {
                'title': simulation.title,
                'job_id': simulation.job_id,
                'start_time': simulation.start_time,
                # submission time, when user send the request to the backend - jobs may start much later than that
                'end_time': simulation.end_time,
                # end time, when the all jobs are finished and results are merged
                'metadata': {
                    'platform': simulation.platform,
                    'server': 'Yaptide',
                    'input_type': simulation.input_type,
                    'sim_type': simulation.sim_type
                }
            } for simulation in simulations
        ],
        'page_count':
        pagination.pages,
        'simulations_count':
        pagination.total,
    }
    return yaptide_response(message='User Simulations', code=200, content=result)

UserUpdate

Bases: Resource

Class responsible for updating the user

Source code in yaptide/routes/user_routes.py 
120
121
122
123
124
125
126
127
128
129
130
class UserUpdate(Resource):
    """Class responsible for updating the user"""

    @staticmethod
    @requires_auth()
    def post(user: UserModel):
        """Updates user with provided parameters"""
        json_data: dict = request.get_json(force=True)
        if not json_data:
            return error_validation_response()
        return yaptide_response(message=f'User {user.username} updated', code=202)

post staticmethod 

post(user)

Updates user with provided parameters

Source code in yaptide/routes/user_routes.py 
123
124
125
126
127
128
129
130
@staticmethod
@requires_auth()
def post(user: UserModel):
    """Updates user with provided parameters"""
    json_data: dict = request.get_json(force=True)
    if not json_data:
        return error_validation_response()
    return yaptide_response(message=f'User {user.username} updated', code=202)