Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
app: App,
state_machine: StateMachine,
state: State,
filter_: Any,
) -> List[str]:
"""Util to get all the labels in an action state that need retrying."""
states_by_rank = app.session.query(
History.label_name,
History.new_state,
func.row_number().over(
# Our model type stubs define the `id` attribute as `int`, yet
# sqlalchemy actually allows the attribute to be used for ordering
# like this; ignore the type check here specifically rather than
# complicate our type definitions.
order_by=History.id.desc(), # type: ignore
partition_by=History.label_name,
).label('rank'),
).filter_by(
label_state_machine=state_machine.name,
).subquery()
ranked_transitions = app.session.query(
states_by_rank.c.label_name,
).filter(
states_by_rank.c.rank == 1,
states_by_rank.c.new_state == state.name,
).join(Label).filter(
filter_,
)
return [x for x, in ranked_transitions]
def _validate_no_labels_in_nonexistent_states(state_machine, app):
states = [x.name for x in state_machine.states]
states_by_rank = app.session.query(
History.label_name,
History.new_state,
func.row_number().over(
order_by=History.id.desc(),
partition_by=History.label_name,
).label('rank'),
).filter_by(
label_state_machine=state_machine.name,
).subquery()
invalid_labels_and_states = app.session.query(
states_by_rank.c.label_name,
states_by_rank.c.new_state,
).filter(
states_by_rank.c.rank == 1,
~(
states_by_rank.c.new_state.in_(states) |
states_by_rank.c.new_state.is_(None)
),
).all()
def get_current_history(app: App, label: LabelRef) -> History:
"""Get a label's last history entry."""
history_entry = app.session.query(History).filter_by(
label_name=label.name,
label_state_machine=label.state_machine,
).order_by(
# Our model type stubs define the `id` attribute as `int`, yet
# sqlalchemy actually allows the attribute to be used for ordering like
# this; ignore the type check here specifically rather than complicate
# our type definitions.
History.id.desc(), # type: ignore
).first()
if history_entry is None:
raise UnknownLabel(label)
return history_entry