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

JobStateField

Bases: Field

custom deserializer for job_state field

Source code in yaptide/routes/user_routes.py 
40
41
42
43
44
45
46
class JobStateField(fields.Field):
    """custom deserializer for job_state field"""

    @staticmethod
    def _deserialize(value, attr, data, **kwargs):
        """deserializes job_state, which is expected to come as comma-separated list of states"""
        return value.split(',') if isinstance(value, str) else []

OrderBy

Bases: Enum

Order by column

Source code in yaptide/routes/user_routes.py 
27
28
29
30
31
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 
20
21
22
23
24
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 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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)
        job_state = JobStateField(validate=validate_job_state, load_default=[])

    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)
        if len(params_dict['job_state']) > 0:
            query = query.filter(SimulationModel.job_state.in_(params_dict['job_state']))
        query = query.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 
61
62
63
64
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 
52
53
54
55
56
57
58
59
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)
    job_state = JobStateField(validate=validate_job_state, load_default=[])
job_state class-attribute instance-attribute 
job_state = JobStateField(
    validate=validate_job_state, load_default=[]
)
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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@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 
 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
@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)
    if len(params_dict['job_state']) > 0:
        query = query.filter(SimulationModel.job_state.in_(params_dict['job_state']))
    query = query.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 
139
140
141
142
143
144
145
146
147
148
149
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 
142
143
144
145
146
147
148
149
@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)

validate_job_state 

validate_job_state(states)

check if all states are correct values of EntityState enum

Source code in yaptide/routes/user_routes.py 
34
35
36
37
def validate_job_state(states: List[str]):
    """check if all states are correct values of EntityState enum"""
    if not set(states).issubset({es.value for es in EntityState}):
        raise ValidationError('Invalid job state')