INNER CODE UNIT · Python

_update_one_launch_plan

flyteorg/flytekit · flytekit/clis/flyte_cli/main.py:252

def _update_one_launch_plan(client: _friendly_client.SynchronousFlyteClient, urn, state):
    if state == "active":
        state = _launch_plan.LaunchPlanState.ACTIVE
    else:
        state = _launch_plan.LaunchPlanState.INACTIVE
    client.update_launch_plan(cli_identifiers.Identifier.from_python_std(urn), state)
    click.echo("Successfully updated {}".format(_tt(urn)))


def _render_schedule_expr(lp):
    sched_expr = "NONE"
    if lp.spec.entity_metadata.schedule and lp.spec.entity_metadata.schedule.cron_expression:
        sched_expr = "cron({cron_expr})".format(cron_expr=_tt(lp.spec.entity_metadata.schedule.cron_expression))
    elif lp.spec.entity_metadata.schedule and lp.spec.entity_metadata.schedule.rate:
        sched_expr = "rate({unit}={value})".format(
            unit=_tt(_Schedule.FixedRateUnit.enum_to_string(lp.spec.entity_metadata.schedule.rate.unit)),
            value=_tt(lp.spec.entity_metadata.schedule.rate.value),
        )

View source record →