Skip to content

db_methods

persistence.db_methods

add_object_to_db 

add_object_to_db(obj, make_commit=True)

Adds object to database and makes commit

Source code in yaptide/persistence/db_methods.py 
12
13
14
15
16
def add_object_to_db(obj: db.Model, make_commit: bool = True) -> None:
    """Adds object to database and makes commit"""
    db.session.add(obj)
    if make_commit:
        make_commit_to_db()

delete_object_from_db 

delete_object_from_db(obj, make_commit=True)

Deletes object from database and makes commit

Source code in yaptide/persistence/db_methods.py 
19
20
21
22
23
def delete_object_from_db(obj: db.Model, make_commit: bool = True) -> None:
    """Deletes object from database and makes commit"""
    db.session.delete(obj)
    if make_commit:
        make_commit_to_db()

fetch_all_clusters 

fetch_all_clusters()

Fetches all clusters

Source code in yaptide/persistence/db_methods.py 
155
156
157
158
def fetch_all_clusters() -> list[ClusterModel]:
    """Fetches all clusters"""
    clusters = db.session.query(ClusterModel).all()
    return clusters

fetch_batch_simulation_by_job_id 

fetch_batch_simulation_by_job_id(job_id)

Fetches batch simulation by job id

Source code in yaptide/persistence/db_methods.py 
72
73
74
75
def fetch_batch_simulation_by_job_id(job_id: str) -> BatchSimulationModel:
    """Fetches batch simulation by job id"""
    simulation = db.session.query(BatchSimulationModel).filter_by(job_id=job_id).first()
    return simulation

fetch_batch_tasks_by_sim_id 

fetch_batch_tasks_by_sim_id(sim_id)

Fetches batch tasks by simulation

Source code in yaptide/persistence/db_methods.py 
112
113
114
115
def fetch_batch_tasks_by_sim_id(sim_id: int) -> list[BatchTaskModel]:
    """Fetches batch tasks by simulation"""
    tasks = db.session.query(BatchTaskModel).filter_by(simulation_id=sim_id).all()
    return tasks

fetch_celery_simulation_by_job_id 

fetch_celery_simulation_by_job_id(job_id)

Fetches celery simulation by job id

Source code in yaptide/persistence/db_methods.py 
66
67
68
69
def fetch_celery_simulation_by_job_id(job_id: str) -> CelerySimulationModel:
    """Fetches celery simulation by job id"""
    simulation = db.session.query(CelerySimulationModel).filter_by(job_id=job_id).first()
    return simulation

fetch_celery_tasks_by_sim_id 

fetch_celery_tasks_by_sim_id(sim_id)

Fetches celery tasks by simulation

Source code in yaptide/persistence/db_methods.py 
106
107
108
109
def fetch_celery_tasks_by_sim_id(sim_id: int) -> list[CeleryTaskModel]:
    """Fetches celery tasks by simulation"""
    tasks = db.session.query(CeleryTaskModel).filter_by(simulation_id=sim_id).all()
    return tasks

fetch_cluster_by_id 

fetch_cluster_by_id(cluster_id)

Fetches cluster by id

Source code in yaptide/persistence/db_methods.py 
161
162
163
164
def fetch_cluster_by_id(cluster_id: int) -> ClusterModel:
    """Fetches cluster by id"""
    cluster = db.session.query(ClusterModel).filter_by(id=cluster_id).first()
    return cluster

fetch_estimator_by_sim_id_and_est_name 

fetch_estimator_by_sim_id_and_est_name(sim_id, est_name)

Fetches estimator by simulation id and estimator name

Source code in yaptide/persistence/db_methods.py 
137
138
139
140
def fetch_estimator_by_sim_id_and_est_name(sim_id: int, est_name: str) -> EstimatorModel:
    """Fetches estimator by simulation id and estimator name"""
    estimator = db.session.query(EstimatorModel).filter_by(simulation_id=sim_id, name=est_name).first()
    return estimator

fetch_estimator_names_by_job_id 

fetch_estimator_names_by_job_id(job_id)

Fetches estimators names by job id Returns a list of estimator names if the simulation exists, or None if no simulation is found for the provided job ID.

Source code in yaptide/persistence/db_methods.py 
124
125
126
127
128
129
130
131
132
133
134
def fetch_estimator_names_by_job_id(job_id: int) -> Optional[list[str]]:
    """Fetches estimators names by job id
    Returns a list of estimator names if the simulation exists,
    or None if no simulation is found for the provided job ID.
    """
    simulation_id = fetch_simulation_id_by_job_id(job_id=job_id)
    if not simulation_id:
        return None
    estimator_names_tuples = db.session.query(EstimatorModel.name).filter_by(simulation_id=simulation_id).all()
    estimator_names = [name for (name, ) in estimator_names_tuples]
    return estimator_names

fetch_estimators_by_sim_id 

fetch_estimators_by_sim_id(sim_id)

Fetches estimators by simulation id

Source code in yaptide/persistence/db_methods.py 
118
119
120
121
def fetch_estimators_by_sim_id(sim_id: int) -> list[EstimatorModel]:
    """Fetches estimators by simulation id"""
    estimators = db.session.query(EstimatorModel).filter_by(simulation_id=sim_id).all()
    return estimators

fetch_input_by_sim_id 

fetch_input_by_sim_id(sim_id)

Fetches input by simulation id

Source code in yaptide/persistence/db_methods.py 
167
168
169
170
def fetch_input_by_sim_id(sim_id: int) -> InputModel:
    """Fetches input by simulation id"""
    input_model = db.session.query(InputModel).filter_by(simulation_id=sim_id).first()
    return input_model

fetch_keycloak_user_by_username 

fetch_keycloak_user_by_username(username)

Fetches user by username

Source code in yaptide/persistence/db_methods.py 
44
45
46
47
def fetch_keycloak_user_by_username(username: str) -> KeycloakUserModel:
    """Fetches user by username"""
    user = db.session.query(KeycloakUserModel).filter_by(username=username).first()
    return user

fetch_logfiles_by_sim_id 

fetch_logfiles_by_sim_id(sim_id)

Fetches logfiles by simulation id

Source code in yaptide/persistence/db_methods.py 
173
174
175
176
def fetch_logfiles_by_sim_id(sim_id: int) -> LogfilesModel:
    """Fetches logfiles by simulation id"""
    logfiles = db.session.query(LogfilesModel).filter_by(simulation_id=sim_id).first()
    return logfiles

fetch_page_by_est_id_and_page_number 

fetch_page_by_est_id_and_page_number(est_id, page_number)

Fetches page by estimator id and page number

Source code in yaptide/persistence/db_methods.py 
149
150
151
152
def fetch_page_by_est_id_and_page_number(est_id: int, page_number: int) -> PageModel:
    """Fetches page by estimator id and page number"""
    page = db.session.query(PageModel).filter_by(estimator_id=est_id, page_number=page_number).first()
    return page

fetch_pages_by_estimator_id 

fetch_pages_by_estimator_id(est_id)

Fetches pages by estimator id

Source code in yaptide/persistence/db_methods.py 
143
144
145
146
def fetch_pages_by_estimator_id(est_id: int) -> list[PageModel]:
    """Fetches pages by estimator id"""
    pages = db.session.query(PageModel).filter_by(estimator_id=est_id).all()
    return pages

fetch_simulation_by_job_id 

fetch_simulation_by_job_id(job_id)

Fetches simulation by job id

Source code in yaptide/persistence/db_methods.py 
50
51
52
53
54
def fetch_simulation_by_job_id(job_id: str) -> Union[BatchSimulationModel, CelerySimulationModel]:
    """Fetches simulation by job id"""
    SimulationPoly = with_polymorphic(SimulationModel, [BatchSimulationModel, CelerySimulationModel])
    simulation = db.session.query(SimulationPoly).filter_by(job_id=job_id).first()
    return simulation

fetch_simulation_by_sim_id 

fetch_simulation_by_sim_id(sim_id)

Fetches simulation by sim id

Source code in yaptide/persistence/db_methods.py 
78
79
80
81
82
def fetch_simulation_by_sim_id(sim_id: int) -> Union[BatchSimulationModel, CelerySimulationModel]:
    """Fetches simulation by sim id"""
    SimulationPoly = with_polymorphic(SimulationModel, [BatchSimulationModel, CelerySimulationModel])
    simulation = db.session.query(SimulationPoly).filter_by(id=sim_id).first()
    return simulation

fetch_simulation_id_by_job_id 

fetch_simulation_id_by_job_id(job_id)

Fetches simulation_id by job_id for both Celery and Batch simulations. Returns simulation_id if simulation exists, or None if no simulation is found.

Source code in yaptide/persistence/db_methods.py 
57
58
59
60
61
62
63
def fetch_simulation_id_by_job_id(job_id: str) -> Optional[int]:
    """Fetches simulation_id by job_id for both Celery and Batch simulations.
    Returns simulation_id if simulation exists,
    or None if no simulation is found.
    """
    simulation_id = db.session.query(SimulationModel.id).filter_by(job_id=job_id).first()
    return simulation_id[0] if simulation_id else None

fetch_simulations_by_user_id 

fetch_simulations_by_user_id(user_id)

Fetches simulations by user id

Source code in yaptide/persistence/db_methods.py 
85
86
87
88
89
def fetch_simulations_by_user_id(user_id: int) -> Union[list[BatchSimulationModel], list[CelerySimulationModel]]:
    """Fetches simulations by user id"""
    SimulationPoly = with_polymorphic(SimulationModel, [BatchSimulationModel, CelerySimulationModel])
    simulations = db.session.query(SimulationPoly).filter_by(user_id=user_id).all()
    return simulations

fetch_task_by_sim_id_and_task_id 

fetch_task_by_sim_id_and_task_id(sim_id, task_id)

Fetches task by simulation id and task id

Source code in yaptide/persistence/db_methods.py 
92
93
94
95
96
def fetch_task_by_sim_id_and_task_id(sim_id: int, task_id: str) -> Union[BatchTaskModel, CeleryTaskModel]:
    """Fetches task by simulation id and task id"""
    TaskPoly = with_polymorphic(TaskModel, [BatchTaskModel, CeleryTaskModel])
    task = db.session.query(TaskPoly).filter_by(simulation_id=sim_id, task_id=task_id).first()
    return task

fetch_tasks_by_sim_id 

fetch_tasks_by_sim_id(sim_id)

Fetches tasks by simulation id

Source code in yaptide/persistence/db_methods.py 
 99
100
101
102
103
def fetch_tasks_by_sim_id(sim_id: int) -> Union[list[BatchTaskModel], list[CeleryTaskModel]]:
    """Fetches tasks by simulation id"""
    TaskPoly = with_polymorphic(TaskModel, [BatchTaskModel, CeleryTaskModel])
    tasks = db.session.query(TaskPoly).filter_by(simulation_id=sim_id).all()
    return tasks

fetch_user_by_id 

fetch_user_by_id(user_id)

Fetches user by id

Source code in yaptide/persistence/db_methods.py 
31
32
33
34
35
def fetch_user_by_id(user_id: int) -> Union[KeycloakUserModel, YaptideUserModel]:
    """Fetches user by id"""
    UserPoly = with_polymorphic(UserModel, [YaptideUserModel, KeycloakUserModel])
    user = db.session.query(UserPoly).filter_by(id=user_id).first()
    return user

fetch_yaptide_user_by_username 

fetch_yaptide_user_by_username(username)

Fetches user by username

Source code in yaptide/persistence/db_methods.py 
38
39
40
41
def fetch_yaptide_user_by_username(username: str) -> YaptideUserModel:
    """Fetches user by username"""
    user = db.session.query(YaptideUserModel).filter_by(username=username).first()
    return user

make_commit_to_db 

make_commit_to_db()

Makes commit

Source code in yaptide/persistence/db_methods.py 
26
27
28
def make_commit_to_db():
    """Makes commit"""
    db.session.commit()

update_simulation_state 

update_simulation_state(simulation, update_dict)

Updates simulation state and makes commit

Source code in yaptide/persistence/db_methods.py 
185
186
187
188
189
190
def update_simulation_state(simulation: Union[BatchSimulationModel, CelerySimulationModel], update_dict: dict) -> None:
    """Updates simulation state and makes commit"""
    if simulation.update_state(update_dict):
        db.session.commit()
    else:
        logging.warning("Simulation state not updated, skipping commit")

update_task_state 

update_task_state(task, update_dict)

Updates task state and makes commit

Source code in yaptide/persistence/db_methods.py 
179
180
181
182
def update_task_state(task: Union[BatchTaskModel, CeleryTaskModel], update_dict: dict) -> None:
    """Updates task state and makes commit"""
    task.update_state(update_dict)
    db.session.commit()