Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
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
"""State machine exceptions."""
class UnknownLabel(ValueError):
"""Represents a label unknown in the given state machine."""
deleted = False
class DeletedLabel(UnknownLabel):
"""Represents a label deleted in the given state machine."""
deleted = True
class UnknownStateMachine(ValueError):
"""Represents a state machine not in the system."""
class LabelAlreadyExists(ValueError):
"""Thrown when a label already exists in the state machine."""
def lock_label(app: App, label: LabelRef) -> Label:
"""Lock a label in the current transaction."""
row = app.session.query(Label).filter_by(
name=label.name,
state_machine=label.state_machine,
).with_for_update().first()
if row is None:
raise UnknownLabel(label)
return row