prefect.orion.schemas.sorting
¶
Schemas for sorting Orion API objects.
FlowRunSort
¶
Defines flow run sorting options.
Source code in prefect/orion/schemas/sorting.py
class FlowRunSort(AutoEnum):
"""Defines flow run sorting options."""
ID_DESC = AutoEnum.auto()
EXPECTED_START_TIME_ASC = AutoEnum.auto()
EXPECTED_START_TIME_DESC = AutoEnum.auto()
NAME_ASC = AutoEnum.auto()
NAME_DESC = AutoEnum.auto()
NEXT_SCHEDULED_START_TIME_ASC = AutoEnum.auto()
END_TIME_DESC = AutoEnum.auto()
def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
"""Return an expression used to sort flow runs"""
sort_mapping = {
"ID_DESC": db.FlowRun.id.desc(),
"EXPECTED_START_TIME_ASC": db.FlowRun.expected_start_time.asc(),
"EXPECTED_START_TIME_DESC": db.FlowRun.expected_start_time.desc(),
"NAME_ASC": db.FlowRun.name.asc(),
"NAME_DESC": db.FlowRun.name.desc(),
"NEXT_SCHEDULED_START_TIME_ASC": db.FlowRun.next_scheduled_start_time.asc(),
"END_TIME_DESC": db.FlowRun.end_time.desc(),
}
return sort_mapping[self.value]
FlowSort
¶
Defines flow sorting options.
Source code in prefect/orion/schemas/sorting.py
class FlowSort(AutoEnum):
"""Defines flow sorting options."""
CREATED_DESC = AutoEnum.auto()
UPDATED_DESC = AutoEnum.auto()
NAME_ASC = AutoEnum.auto()
NAME_DESC = AutoEnum.auto()
def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
"""Return an expression used to sort flow runs"""
sort_mapping = {
"CREATED_DESC": db.Flow.created.desc(),
"UPDATED_DESC": db.Flow.updated.desc(),
"NAME_ASC": db.Flow.name.asc(),
"NAME_DESC": db.Flow.name.desc(),
}
return sort_mapping[self.value]
LogSort
¶
Defines log sorting options.
Source code in prefect/orion/schemas/sorting.py
class LogSort(AutoEnum):
"""Defines log sorting options."""
TIMESTAMP_ASC = AutoEnum.auto()
TIMESTAMP_DESC = AutoEnum.auto()
LEVEL_ASC = AutoEnum.auto()
LEVEL_DESC = AutoEnum.auto()
FLOW_RUN_ID_ASC = AutoEnum.auto()
FLOW_RUN_ID_DESC = AutoEnum.auto()
TASK_RUN_ID_ASC = AutoEnum.auto()
TASK_RUN_ID_DESC = AutoEnum.auto()
def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
"""Return an expression used to sort task runs"""
sort_mapping = {
"TIMESTAMP_ASC": db.Log.timestamp.asc(),
"TIMESTAMP_DESC": db.Log.timestamp.desc(),
"LEVEL_ASC": db.Log.level.asc(),
"LEVEL_DESC": db.Log.level.desc(),
"FLOW_RUN_ID_ASC": db.Log.flow_run_id.asc(),
"FLOW_RUN_ID_DESC": db.Log.flow_run_id.desc(),
"TASK_RUN_ID_ASC": db.Log.task_run_id.asc(),
"TASK_RUN_ID_DESC": db.Log.task_run_id.desc(),
}
return sort_mapping[self.value]
TaskRunSort
¶
Defines task run sorting options.
Source code in prefect/orion/schemas/sorting.py
class TaskRunSort(AutoEnum):
"""Defines task run sorting options."""
ID_DESC = AutoEnum.auto()
EXPECTED_START_TIME_ASC = AutoEnum.auto()
EXPECTED_START_TIME_DESC = AutoEnum.auto()
NAME_ASC = AutoEnum.auto()
NAME_DESC = AutoEnum.auto()
NEXT_SCHEDULED_START_TIME_ASC = AutoEnum.auto()
END_TIME_DESC = AutoEnum.auto()
def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
"""Return an expression used to sort task runs"""
sort_mapping = {
"ID_DESC": db.TaskRun.id.desc(),
"EXPECTED_START_TIME_ASC": db.TaskRun.expected_start_time.asc(),
"EXPECTED_START_TIME_DESC": db.TaskRun.expected_start_time.desc(),
"NAME_ASC": db.TaskRun.name.asc(),
"NAME_DESC": db.TaskRun.name.desc(),
"NEXT_SCHEDULED_START_TIME_ASC": db.TaskRun.next_scheduled_start_time.asc(),
"END_TIME_DESC": db.TaskRun.end_time.desc(),
}
return sort_mapping[self.value]