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 
13
14
15
16
17
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 
20
21
22
23
24
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 
182
183
184
185
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 
73
74
75
76
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 
113
114
115
116
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 
67
68
69
70
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 
107
108
109
110
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 
188
189
190
191
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 
138
139
140
141
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_by_sim_id_and_file_name 

fetch_estimator_by_sim_id_and_file_name(sim_id, file_name)

Fetches estimator by simulation id and estimator name

Source code in yaptide/persistence/db_methods.py 
144
145
146
147
def fetch_estimator_by_sim_id_and_file_name(sim_id: int, file_name: str) -> EstimatorModel:
    """Fetches estimator by simulation id and estimator name"""
    estimator = db.session.query(EstimatorModel).filter_by(simulation_id=sim_id, file_name=file_name).first()
    return estimator

fetch_estimator_id_by_sim_id_and_est_name 

fetch_estimator_id_by_sim_id_and_est_name(sim_id, est_name)

Fetches estimator_id by simulation id and estimator name

Source code in yaptide/persistence/db_methods.py 
150
151
152
153
def fetch_estimator_id_by_sim_id_and_est_name(sim_id: int, est_name: str) -> Optional[int]:
    """Fetches estimator_id by simulation id and estimator name"""
    estimator_id = db.session.query(EstimatorModel.id).filter_by(simulation_id=sim_id, name=est_name).first()
    return estimator_id[0] if estimator_id else None

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 
125
126
127
128
129
130
131
132
133
134
135
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 
119
120
121
122
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 
194
195
196
197
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 
45
46
47
48
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 
200
201
202
203
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 
162
163
164
165
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_est_id_and_page_numbers 

fetch_pages_by_est_id_and_page_numbers(
    est_id, page_numbers
)

Fetches page by estimator id and page number

Source code in yaptide/persistence/db_methods.py 
168
169
170
171
172
def fetch_pages_by_est_id_and_page_numbers(est_id: int, page_numbers: list) -> PageModel:
    """Fetches page by estimator id and page number"""
    pages = db.session.query(PageModel).filter(
        and_(PageModel.estimator_id == est_id, PageModel.page_number.in_(page_numbers))).all()
    return pages

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 
156
157
158
159
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_pages_metadata_by_est_id 

fetch_pages_metadata_by_est_id(est_id)

Fetches estimator by simulation id and estimator name

Source code in yaptide/persistence/db_methods.py 
175
176
177
178
179
def fetch_pages_metadata_by_est_id(est_id: str) -> EstimatorModel:
    """Fetches estimator by simulation id and estimator name"""
    pages_metadata = db.session.query(PageModel.page_number, PageModel.page_name,
                                      PageModel.page_dimension).filter_by(estimator_id=est_id).all()
    return pages_metadata

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 
51
52
53
54
55
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 
79
80
81
82
83
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 
58
59
60
61
62
63
64
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 
86
87
88
89
90
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 
93
94
95
96
97
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 
100
101
102
103
104
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 
32
33
34
35
36
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 
39
40
41
42
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 
27
28
29
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 
212
213
214
215
216
217
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 
206
207
208
209
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()