(
session,
task_id,
direction = 'next'
)
| 277 | return task |
| 278 | |
| 279 | def navigate_tasks_relative_to_given_task( |
| 280 | session, |
| 281 | task_id, |
| 282 | direction = 'next' |
| 283 | ): |
| 284 | |
| 285 | known_task = Task.get_by_id(session, task_id = task_id) |
| 286 | |
| 287 | query = session.query(Task).filter( |
| 288 | Task.status != 'archived', |
| 289 | Task.job_id == known_task.job_id) |
| 290 | |
| 291 | if direction == 'next': |
| 292 | query = query.filter(Task.time_created > known_task.time_created) |
| 293 | query = query.order_by(Task.time_created) |
| 294 | |
| 295 | elif direction == 'previous': |
| 296 | query = query.filter(Task.time_created < known_task.time_created) |
| 297 | query = query.order_by(Task.time_created.desc()) |
| 298 | |
| 299 | discovered_task = query.first() |
| 300 | return discovered_task |
| 301 | |
| 302 | def get_last_task( |
| 303 | session, |
no test coverage detected