Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_label(app: App, label: LabelRef, metadata: Metadata) -> Metadata:
"""Creates a label and starts it in a state machine."""
state_machine = get_state_machine(app, label)
if app.session.query(
app.session.query(Label).filter_by(
name=label.name,
state_machine=label.state_machine,
).exists(),
).scalar():
raise LabelAlreadyExists(label)
app.session.add(
Label(
name=label.name,
state_machine=state_machine.name,
metadata=metadata,
history=[
History(
old_state=None,
new_state=state_machine.states[0].name,
),
],
),
)
process_transitions(app, label)
return metadata
def get_label_metadata(
app: App,
label: LabelRef,
state_machine: StateMachine,
) -> Tuple[Dict[str, Any], bool]:
"""Get the metadata and whether the label has been deleted."""
return app.session.query(Label.metadata, Label.deleted).filter_by(
name=label.name,
state_machine=state_machine.name,
).first()
# 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 list_labels(app: App, state_machine: StateMachine) -> Iterable[LabelRef]:
"""
Returns a sorted iterable of labels associated with a state machine.
Labels are returned ordered alphabetically by name.
"""
for (name,) in app.session.query(Label.name).filter_by(
state_machine=state_machine.name,
deleted=False,
):
yield LabelRef(name=name, state_machine=state_machine.name)
can_exit = gate.exit_condition.run(context)
if not can_exit:
return False
destination = choose_next_state(state_machine, gate, context)
app.session.add(History(
label_state_machine=state_machine.name,
label_name=label.name,
created=func.now(),
old_state=gate.name,
new_state=destination.name,
))
app.session.query(Label).filter_by(
name=label.name,
state_machine=label.state_machine,
).update({
'metadata_triggers_processed': True,
})
return True