How to use the routemaster.db.Label function in routemaster

To help you get started, we’ve selected a few routemaster examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github thread / routemaster / routemaster / state_machine / api.py View on Github external
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
github thread / routemaster / routemaster / state_machine / utils.py View on Github external
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()
github thread / routemaster / routemaster / state_machine / utils.py View on Github external
# 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]
github thread / routemaster / routemaster / state_machine / api.py View on Github external
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)
github thread / routemaster / routemaster / state_machine / gates.py View on Github external
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