Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def update_metadata_for_label(
app: App,
label: LabelRef,
update: Metadata,
) -> Metadata:
"""
Updates the metadata for a label.
Moves the label through the state machine as appropriate.
"""
state_machine = get_state_machine(app, label)
needs_gate_evaluation = False
row = lock_label(app, label)
existing_metadata, deleted = row.metadata, row.deleted
if deleted:
raise DeletedLabel(label)
needs_gate_evaluation, current_state = \
needs_gate_evaluation_for_metadata_change(
app,
state_machine,
label,
update,
)
def get_label_state(app: App, label: LabelRef) -> Optional[State]:
"""Finds the current state of a label."""
state_machine = get_state_machine(app, label)
return get_current_state(app, label, state_machine)
def process_transitions(app: App, label: LabelRef) -> None:
"""
Process each transition for a label until it cannot move any further.
Will silently accept DeletedLabel exceptions and end the processing of
transitions.
"""
state_machine = get_state_machine(app, label)
could_progress = True
num_transitions = 0
def _transition() -> bool:
with app.session.begin_nested():
lock_label(app, label)
current_state = get_current_state(app, label, state_machine)
if isinstance(current_state, Action):
return process_action(
app=app,
state=current_state,
state_machine=state_machine,
label=label,
)
def get_label_metadata(app: App, label: LabelRef) -> Metadata:
"""Returns the metadata associated with a label."""
state_machine = get_state_machine(app, label)
row = get_label_metadata_internal(app, label, state_machine)
if row is None:
raise UnknownLabel(label)
metadata, deleted = row
if deleted:
raise DeletedLabel(label)
return metadata
Process a label in a gate, continuing if necessary.
Assumes that `gate` is the current state of the label, and that the label
has been locked.
Returns whether the label progressed in the state machine, for which `True`
implies further progression should be attempted.
"""
if not isinstance(state, Gate): # pragma: no branch
raise ValueError( # pragma: no cover
f"process_gate called with {state.name} which is not an Gate",
)
gate = state
state_machine = get_state_machine(app, label)
metadata, deleted = get_label_metadata(app, label, state_machine)
if deleted:
raise DeletedLabel(label)
history_entry = get_current_history(app, label)
context = context_for_label(
label,
metadata,
state_machine,
gate,
history_entry,
app.logger,
)
can_exit = gate.exit_condition.run(context)
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(
def delete_label(app: App, label: LabelRef) -> None:
"""
Deletes the metadata for a label and marks the label as deleted.
The history for the label is not changed (in order to allow post-hoc
analysis of the path the label took through the state machine).
"""
state_machine = get_state_machine(app, label) # Raises UnknownStateMachine
try:
row = lock_label(app, label)
except UnknownLabel:
return
if row is None or row.deleted:
return
# Record the label as having been deleted and remove its metadata
row.metadata = {}
row.deleted = True
# Add a history entry for the deletion
current_state = get_current_state(app, label, state_machine)