prefect.cli.deployment

Command line interface for working with deployments.

clear_schedules(deployment_name, assume_yes=typer.Option(False, '--accept-yes', '-y', help='Accept the confirmation prompt without prompting')) async

Clear all schedules for a deployment.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("clear")
async def clear_schedules(
    deployment_name: str,
    assume_yes: Optional[bool] = typer.Option(
        False,
        "--accept-yes",
        "-y",
        help="Accept the confirmation prompt without prompting",
    ),
):
    """
    Clear all schedules for a deployment.
    """
    assert_deployment_name_format(deployment_name)
    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(deployment_name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {deployment_name!r} not found!")

        await client.read_flow(deployment.flow_id)

        # Get input from user: confirm removal of all schedules
        if not assume_yes and not typer.confirm(
            "Are you sure you want to clear all schedules for this deployment?",
        ):
            exit_with_error("Clearing schedules cancelled.")

        for schedule in deployment.schedules:
            try:
                await client.delete_deployment_schedule(deployment.id, schedule.id)
            except ObjectNotFound:
                pass

        exit_with_success(f"Cleared all schedules for deployment {deployment_name}")

create_schedule(name, interval=typer.Option(None, '--interval', help='An interval to schedule on, specified in seconds', min=0.0001), interval_anchor=typer.Option(None, '--anchor-date', help='The anchor date for an interval schedule'), rrule_string=typer.Option(None, '--rrule', help='Deployment schedule rrule string'), cron_string=typer.Option(None, '--cron', help='Deployment schedule cron string'), cron_day_or=typer.Option(None, '--day_or', help='Control how croniter handles `day` and `day_of_week` entries'), timezone=typer.Option(None, '--timezone', help="Deployment schedule timezone string e.g. 'America/New_York'"), active=typer.Option(True, '--active', help='Whether the schedule is active. Defaults to True.'), replace=typer.Option(False, '--replace', help="Replace the deployment's current schedule(s) with this new schedule."), assume_yes=typer.Option(False, '--accept-yes', '-y', help='Accept the confirmation prompt without prompting')) async

Create a schedule for a given deployment.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("create")
async def create_schedule(
    name: str,
    interval: Optional[float] = typer.Option(
        None,
        "--interval",
        help="An interval to schedule on, specified in seconds",
        min=0.0001,
    ),
    interval_anchor: Optional[str] = typer.Option(
        None,
        "--anchor-date",
        help="The anchor date for an interval schedule",
    ),
    rrule_string: Optional[str] = typer.Option(
        None, "--rrule", help="Deployment schedule rrule string"
    ),
    cron_string: Optional[str] = typer.Option(
        None, "--cron", help="Deployment schedule cron string"
    ),
    cron_day_or: Optional[str] = typer.Option(
        None,
        "--day_or",
        help="Control how croniter handles `day` and `day_of_week` entries",
    ),
    timezone: Optional[str] = typer.Option(
        None,
        "--timezone",
        help="Deployment schedule timezone string e.g. 'America/New_York'",
    ),
    active: Optional[bool] = typer.Option(
        True,
        "--active",
        help="Whether the schedule is active. Defaults to True.",
    ),
    replace: Optional[bool] = typer.Option(
        False,
        "--replace",
        help="Replace the deployment's current schedule(s) with this new schedule.",
    ),
    assume_yes: Optional[bool] = typer.Option(
        False,
        "--accept-yes",
        "-y",
        help="Accept the confirmation prompt without prompting",
    ),
):
    """
    Create a schedule for a given deployment.
    """
    assert_deployment_name_format(name)

    if sum(option is not None for option in [interval, rrule_string, cron_string]) != 1:
        exit_with_error(
            "Exactly one of `--interval`, `--rrule`, or `--cron` must be provided."
        )

    schedule = None

    if interval_anchor and not interval:
        exit_with_error("An anchor date can only be provided with an interval schedule")

    if interval is not None:
        if interval_anchor:
            try:
                pendulum.parse(interval_anchor)
            except ValueError:
                return exit_with_error("The anchor date must be a valid date string.")
        interval_schedule = {
            "interval": interval,
            "anchor_date": interval_anchor,
            "timezone": timezone,
        }
        schedule = IntervalSchedule(
            **{k: v for k, v in interval_schedule.items() if v is not None}
        )

    if cron_string is not None:
        cron_schedule = {
            "cron": cron_string,
            "day_or": cron_day_or,
            "timezone": timezone,
        }
        schedule = CronSchedule(
            **{k: v for k, v in cron_schedule.items() if v is not None}
        )

    if rrule_string is not None:
        # a timezone in the `rrule_string` gets ignored by the RRuleSchedule constructor
        if "TZID" in rrule_string and not timezone:
            exit_with_error(
                "You can provide a timezone by providing a dict with a `timezone` key"
                " to the --rrule option. E.g. {'rrule': 'FREQ=MINUTELY;INTERVAL=5',"
                " 'timezone': 'America/New_York'}.\nAlternatively, you can provide a"
                " timezone by passing in a --timezone argument."
            )
        try:
            schedule = RRuleSchedule(**json.loads(rrule_string))
            if timezone:
                # override timezone if specified via CLI argument
                schedule.timezone = timezone
        except json.JSONDecodeError:
            schedule = RRuleSchedule(rrule=rrule_string, timezone=timezone)

    if schedule is None:
        return exit_with_success(
            "Could not create a valid schedule from the provided options."
        )

    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {name!r} not found!")

        num_schedules = len(deployment.schedules)
        noun = "schedule" if num_schedules == 1 else "schedules"

        if replace and num_schedules > 0:
            if not assume_yes and not typer.confirm(
                f"Are you sure you want to replace {num_schedules} {noun} for {name}?"
            ):
                return exit_with_error("Schedule replacement cancelled.")

            for existing_schedule in deployment.schedules:
                try:
                    await client.delete_deployment_schedule(
                        deployment.id, existing_schedule.id
                    )
                except ObjectNotFound:
                    pass

        await client.create_deployment_schedules(deployment.id, [(schedule, active)])

        if replace and num_schedules > 0:
            exit_with_success(f"Replaced existing deployment {noun} with new schedule!")
        else:
            exit_with_success("Created deployment schedule!")

delete(name=typer.Argument(None, help="A deployed flow's name: <FLOW_NAME>/<DEPLOYMENT_NAME>"), deployment_id=typer.Option(None, '--id', help='A deployment id to search for if no name is given')) async

Delete a deployment.

 Examples:  $ prefect deployment delete test_flow/test_deployment $ prefect deployment delete --id dfd3e220-a130-4149-9af6-8d487e02fea6

Source code in src/prefect/cli/deployment.py

@deployment_app.command()
async def delete(
    name: Optional[str] = typer.Argument(
        None, help="A deployed flow's name: <FLOW_NAME>/<DEPLOYMENT_NAME>"
    ),
    deployment_id: Optional[str] = typer.Option(
        None, "--id", help="A deployment id to search for if no name is given"
    ),
):
    """
    Delete a deployment.

    \b
    Examples:
        \b
        $ prefect deployment delete test_flow/test_deployment
        $ prefect deployment delete --id dfd3e220-a130-4149-9af6-8d487e02fea6
    """
    async with get_client() as client:
        if name is None and deployment_id is not None:
            try:
                if is_interactive() and not typer.confirm(
                    (
                        f"Are you sure you want to delete deployment with id {deployment_id!r}?"
                    ),
                    default=False,
                ):
                    exit_with_error("Deletion aborted.")
                await client.delete_deployment(deployment_id)
                exit_with_success(f"Deleted deployment '{deployment_id}'.")
            except ObjectNotFound:
                exit_with_error(f"Deployment {deployment_id!r} not found!")
        elif name is not None:
            try:
                deployment = await client.read_deployment_by_name(name)
                if is_interactive() and not typer.confirm(
                    (f"Are you sure you want to delete deployment with name {name!r}?"),
                    default=False,
                ):
                    exit_with_error("Deletion aborted.")
                await client.delete_deployment(deployment.id)
                exit_with_success(f"Deleted deployment '{name}'.")
            except ObjectNotFound:
                exit_with_error(f"Deployment {name!r} not found!")
        else:
            exit_with_error("Must provide a deployment name or id")

delete_schedule(deployment_name, schedule_id, assume_yes=typer.Option(False, '--accept-yes', '-y', help='Accept the confirmation prompt without prompting')) async

Delete a deployment schedule.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("delete")
async def delete_schedule(
    deployment_name: str,
    schedule_id: UUID,
    assume_yes: Optional[bool] = typer.Option(
        False,
        "--accept-yes",
        "-y",
        help="Accept the confirmation prompt without prompting",
    ),
):
    """
    Delete a deployment schedule.
    """
    assert_deployment_name_format(deployment_name)

    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(deployment_name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {deployment_name} not found!")

        try:
            schedule = [s for s in deployment.schedules if s.id == schedule_id][0]
        except IndexError:
            return exit_with_error("Deployment schedule not found!")

        if not assume_yes and not typer.confirm(
            f"Are you sure you want to delete this schedule: {schedule.schedule}",
        ):
            return exit_with_error("Deletion cancelled.")

        try:
            await client.delete_deployment_schedule(deployment.id, schedule_id)
        except ObjectNotFound:
            exit_with_error("Deployment schedule not found!")

        exit_with_success(f"Deleted deployment schedule {schedule_id}")

inspect(name) async

View details about a deployment.

 Example:  $ prefect deployment inspect "hello-world/my-deployment" { 'id': '610df9c3-0fb4-4856-b330-67f588d20201', 'created': '2022-08-01T18:36:25.192102+00:00', 'updated': '2022-08-01T18:36:25.188166+00:00', 'name': 'my-deployment', 'description': None, 'flow_id': 'b57b0aa2-ef3a-479e-be49-381fb0483b4e', 'schedules': None, 'parameters': {'name': 'Marvin'}, 'tags': ['test'], 'parameter_openapi_schema': { 'title': 'Parameters', 'type': 'object', 'properties': { 'name': { 'title': 'name', 'type': 'string' } }, 'required': ['name'] }, 'storage_document_id': '63ef008f-1e5d-4e07-a0d4-4535731adb32', 'infrastructure_document_id': '6702c598-7094-42c8-9785-338d2ec3a028', 'infrastructure': { 'type': 'process', 'env': {}, 'labels': {}, 'name': None, 'command': ['python', '-m', 'prefect.engine'], 'stream_output': True } }

Source code in src/prefect/cli/deployment.py

@deployment_app.command()
async def inspect(name: str):
    """
    View details about a deployment.

    \b
    Example:
        \b
        $ prefect deployment inspect "hello-world/my-deployment"
        {
            'id': '610df9c3-0fb4-4856-b330-67f588d20201',
            'created': '2022-08-01T18:36:25.192102+00:00',
            'updated': '2022-08-01T18:36:25.188166+00:00',
            'name': 'my-deployment',
            'description': None,
            'flow_id': 'b57b0aa2-ef3a-479e-be49-381fb0483b4e',
            'schedules': None,
            'parameters': {'name': 'Marvin'},
            'tags': ['test'],
            'parameter_openapi_schema': {
                'title': 'Parameters',
                'type': 'object',
                'properties': {
                    'name': {
                        'title': 'name',
                        'type': 'string'
                    }
                },
                'required': ['name']
            },
            'storage_document_id': '63ef008f-1e5d-4e07-a0d4-4535731adb32',
            'infrastructure_document_id': '6702c598-7094-42c8-9785-338d2ec3a028',
            'infrastructure': {
                'type': 'process',
                'env': {},
                'labels': {},
                'name': None,
                'command': ['python', '-m', 'prefect.engine'],
                'stream_output': True
            }
        }

    """
    assert_deployment_name_format(name)

    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(name)
        except ObjectNotFound:
            exit_with_error(f"Deployment {name!r} not found!")

        deployment_json = deployment.model_dump(mode="json")

        if deployment.infrastructure_document_id:
            deployment_json["infrastructure"] = Block._from_block_document(
                await client.read_block_document(deployment.infrastructure_document_id)
            ).model_dump(
                exclude={"_block_document_id", "_block_document_name", "_is_anonymous"}
            )

        deployment_json["automations"] = [
            a.model_dump()
            for a in await client.read_resource_related_automations(
                f"prefect.deployment.{deployment.id}"
            )
        ]

    app.console.print(Pretty(deployment_json))

list_schedules(deployment_name) async

View all schedules for a deployment.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("ls")
async def list_schedules(deployment_name: str):
    """
    View all schedules for a deployment.
    """
    assert_deployment_name_format(deployment_name)
    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(deployment_name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {deployment_name!r} not found!")

    def sort_by_created_key(schedule: DeploymentScheduleCreate):  # type: ignore
        assert schedule.created is not None, "All schedules should have a created time."
        return pendulum.now("utc") - schedule.created

    def schedule_details(schedule: DeploymentScheduleCreate) -> str:
        if isinstance(schedule.schedule, IntervalSchedule):
            return f"interval: {schedule.schedule.interval}s"
        elif isinstance(schedule.schedule, CronSchedule):
            return f"cron: {schedule.schedule.cron}"
        elif isinstance(schedule.schedule, RRuleSchedule):
            return f"rrule: {schedule.schedule.rrule}"
        else:
            return "unknown"

    table = Table(
        title="Deployment Schedules",
    )
    table.add_column("ID", style="blue", no_wrap=True)
    table.add_column("Schedule", style="cyan", no_wrap=False)
    table.add_column("Active", style="purple", no_wrap=True)

    for schedule in sorted(deployment.schedules, key=sort_by_created_key):
        table.add_row(
            str(schedule.id),
            schedule_details(schedule),
            str(schedule.active),
        )

    app.console.print(table)

ls(flow_name=None, by_created=False) async

View all deployments or deployments for specific flows.

Source code in src/prefect/cli/deployment.py

@deployment_app.command()
async def ls(flow_name: Optional[List[str]] = None, by_created: bool = False):
    """
    View all deployments or deployments for specific flows.
    """
    async with get_client() as client:
        deployments = await client.read_deployments(
            flow_filter=FlowFilter(name={"any_": flow_name}) if flow_name else None
        )
        flows = {
            flow.id: flow
            for flow in await client.read_flows(
                flow_filter=FlowFilter(id={"any_": [d.flow_id for d in deployments]})
            )
        }

    def sort_by_name_keys(d):
        return flows[d.flow_id].name, d.name

    def sort_by_created_key(d):
        return pendulum.now("utc") - d.created

    table = Table(
        title="Deployments",
    )
    table.add_column("Name", style="blue", no_wrap=True)
    table.add_column("ID", style="cyan", no_wrap=True)

    for deployment in sorted(
        deployments, key=sort_by_created_key if by_created else sort_by_name_keys
    ):
        table.add_row(
            f"{flows[deployment.flow_id].name}/[bold]{deployment.name}[/]",
            str(deployment.id),
        )

    app.console.print(table)

pause_schedule(deployment_name, schedule_id) async

Pause a deployment schedule.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("pause")
async def pause_schedule(deployment_name: str, schedule_id: UUID):
    """
    Pause a deployment schedule.
    """
    assert_deployment_name_format(deployment_name)

    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(deployment_name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {deployment_name!r} not found!")

        try:
            schedule = [s for s in deployment.schedules if s.id == schedule_id][0]
        except IndexError:
            return exit_with_error("Deployment schedule not found!")

        if not schedule.active:
            return exit_with_error(
                f"Deployment schedule {schedule_id} is already inactive"
            )

        await client.update_deployment_schedule(
            deployment.id, schedule_id, active=False
        )
        exit_with_success(
            f"Paused schedule {schedule.schedule} for deployment {deployment_name}"
        )

resume_schedule(deployment_name, schedule_id) async

Resume a deployment schedule.

Source code in src/prefect/cli/deployment.py

@schedule_app.command("resume")
async def resume_schedule(deployment_name: str, schedule_id: UUID):
    """
    Resume a deployment schedule.
    """
    assert_deployment_name_format(deployment_name)

    async with get_client() as client:
        try:
            deployment = await client.read_deployment_by_name(deployment_name)
        except ObjectNotFound:
            return exit_with_error(f"Deployment {deployment_name!r} not found!")

        try:
            schedule = [s for s in deployment.schedules if s.id == schedule_id][0]
        except IndexError:
            return exit_with_error("Deployment schedule not found!")

        if schedule.active:
            return exit_with_error(
                f"Deployment schedule {schedule_id} is already active"
            )

        await client.update_deployment_schedule(deployment.id, schedule_id, active=True)
        exit_with_success(
            f"Resumed schedule {schedule.schedule} for deployment {deployment_name}"
        )

run(name=typer.Argument(None, help="A deployed flow's name: <FLOW_NAME>/<DEPLOYMENT_NAME>"), deployment_id=typer.Option(None, '--id', help='A deployment id to search for if no name is given'), job_variables=typer.Option(None, '-jv', '--job-variable', help='A key, value pair (key=value) specifying a flow run job variable. The value will be interpreted as JSON. May be passed multiple times to specify multiple job variable values.'), params=typer.Option(None, '-p', '--param', help='A key, value pair (key=value) specifying a flow parameter. The value will be interpreted as JSON. May be passed multiple times to specify multiple parameter values.'), multiparams=typer.Option(None, '--params', help="A mapping of parameters to values. To use a stdin, pass '-'. Any parameters passed with `--param` will take precedence over these values."), start_in=typer.Option(None, '--start-in', help="A human-readable string specifying a time interval to wait before starting the flow run. E.g. 'in 5 minutes', 'in 1 hour', 'in 2 days'."), start_at=typer.Option(None, '--start-at', help="A human-readable string specifying a time to start the flow run. E.g. 'at 5:30pm', 'at 2022-08-01 17:30', 'at 2022-08-01 17:30:00'."), tags=typer.Option(None, '--tag', help='Tag(s) to be applied to flow run.'), watch=typer.Option(False, '--watch', help='Whether to poll the flow run until a terminal state is reached.'), watch_interval=typer.Option(None, '--watch-interval', help='How often to poll the flow run for state changes (in seconds).'), watch_timeout=typer.Option(None, '--watch-timeout', help='Timeout for --watch.')) async

Create a flow run for the given flow and deployment.

The flow run will be scheduled to run immediately unless --start-in or --start-at is specified. The flow run will not execute until a worker starts. To watch the flow run until it reaches a terminal state, use the --watch flag.

Source code in src/prefect/cli/deployment.py

@deployment_app.command()
async def run(
    name: Optional[str] = typer.Argument(
        None, help="A deployed flow's name: <FLOW_NAME>/<DEPLOYMENT_NAME>"
    ),
    deployment_id: Optional[str] = typer.Option(
        None,
        "--id",
        help=("A deployment id to search for if no name is given"),
    ),
    job_variables: List[str] = typer.Option(
        None,
        "-jv",
        "--job-variable",
        help=(
            "A key, value pair (key=value) specifying a flow run job variable. The value will"
            " be interpreted as JSON. May be passed multiple times to specify multiple"
            " job variable values."
        ),
    ),
    params: List[str] = typer.Option(
        None,
        "-p",
        "--param",
        help=(
            "A key, value pair (key=value) specifying a flow parameter. The value will"
            " be interpreted as JSON. May be passed multiple times to specify multiple"
            " parameter values."
        ),
    ),
    multiparams: Optional[str] = typer.Option(
        None,
        "--params",
        help=(
            "A mapping of parameters to values. To use a stdin, pass '-'. Any "
            "parameters passed with `--param` will take precedence over these values."
        ),
    ),
    start_in: Optional[str] = typer.Option(
        None,
        "--start-in",
        help=(
            "A human-readable string specifying a time interval to wait before starting"
            " the flow run. E.g. 'in 5 minutes', 'in 1 hour', 'in 2 days'."
        ),
    ),
    start_at: Optional[str] = typer.Option(
        None,
        "--start-at",
        help=(
            "A human-readable string specifying a time to start the flow run. E.g."
            " 'at 5:30pm', 'at 2022-08-01 17:30', 'at 2022-08-01 17:30:00'."
        ),
    ),
    tags: List[str] = typer.Option(
        None,
        "--tag",
        help=("Tag(s) to be applied to flow run."),
    ),
    watch: bool = typer.Option(
        False,
        "--watch",
        help=("Whether to poll the flow run until a terminal state is reached."),
    ),
    watch_interval: Optional[int] = typer.Option(
        None,
        "--watch-interval",
        help=("How often to poll the flow run for state changes (in seconds)."),
    ),
    watch_timeout: Optional[int] = typer.Option(
        None,
        "--watch-timeout",
        help=("Timeout for --watch."),
    ),
):
    """
    Create a flow run for the given flow and deployment.

    The flow run will be scheduled to run immediately unless `--start-in` or `--start-at` is specified.
    The flow run will not execute until a worker starts.
    To watch the flow run until it reaches a terminal state, use the `--watch` flag.
    """
    import dateparser

    now = pendulum.now("UTC")

    multi_params = {}
    if multiparams:
        if multiparams == "-":
            multiparams = sys.stdin.read()
            if not multiparams:
                exit_with_error("No data passed to stdin")

        try:
            multi_params = json.loads(multiparams)
        except ValueError as exc:
            exit_with_error(f"Failed to parse JSON: {exc}")
        if watch_interval and not watch:
            exit_with_error(
                "`--watch-interval` can only be used with `--watch`.",
            )
    cli_params = _load_json_key_values(params or [], "parameter")
    conflicting_keys = set(cli_params.keys()).intersection(multi_params.keys())
    if conflicting_keys:
        app.console.print(
            "The following parameters were specified by `--param` and `--params`, the "
            f"`--param` value will be used: {conflicting_keys}"
        )
    parameters = {**multi_params, **cli_params}

    job_vars = _load_json_key_values(job_variables or [], "job variable")
    if start_in and start_at:
        exit_with_error(
            "Only one of `--start-in` or `--start-at` can be set, not both."
        )

    elif start_in is None and start_at is None:
        scheduled_start_time = now
        human_dt_diff = " (now)"
    else:
        if start_in:
            start_time_raw = "in " + start_in
        else:
            start_time_raw = "at " + start_at
        with warnings.catch_warnings():
            # PyTZ throws a warning based on dateparser usage of the library
            # See https://github.com/scrapinghub/dateparser/issues/1089
            warnings.filterwarnings("ignore", module="dateparser")

            try:
                start_time_parsed = dateparser.parse(
                    start_time_raw,
                    settings={
                        "TO_TIMEZONE": "UTC",
                        "RETURN_AS_TIMEZONE_AWARE": False,
                        "PREFER_DATES_FROM": "future",
                        "RELATIVE_BASE": datetime.fromtimestamp(
                            now.timestamp(), tz=timezone.utc
                        ),
                    },
                )

            except Exception as exc:
                exit_with_error(f"Failed to parse '{start_time_raw!r}': {exc!s}")

        if start_time_parsed is None:
            exit_with_error(f"Unable to parse scheduled start time {start_time_raw!r}.")

        scheduled_start_time = pendulum.instance(start_time_parsed)
        human_dt_diff = (
            " (" + pendulum.format_diff(scheduled_start_time.diff(now)) + ")"
        )

    async with get_client() as client:
        deployment = await get_deployment(client, name, deployment_id)
        flow = await client.read_flow(deployment.flow_id)

        deployment_parameters = deployment.parameter_openapi_schema["properties"].keys()
        unknown_keys = set(parameters.keys()).difference(deployment_parameters)
        if unknown_keys:
            available_parameters = (
                (
                    "The following parameters are available on the deployment: "
                    + listrepr(deployment_parameters, sep=", ")
                )
                if deployment_parameters
                else "This deployment does not accept parameters."
            )

            exit_with_error(
                "The following parameters were specified but not found on the "
                f"deployment: {listrepr(unknown_keys, sep=', ')}"
                f"\n{available_parameters}"
            )

        app.console.print(
            f"Creating flow run for deployment '{flow.name}/{deployment.name}'...",
        )

        try:
            flow_run = await client.create_flow_run_from_deployment(
                deployment.id,
                parameters=parameters,
                state=Scheduled(scheduled_time=scheduled_start_time),
                tags=tags,
                job_variables=job_vars,
            )
        except PrefectHTTPStatusError as exc:
            detail = exc.response.json().get("detail")
            if detail:
                exit_with_error(
                    exc.response.json()["detail"],
                )
            else:
                raise

    run_url = urls.url_for(flow_run) or "<no dashboard available>"
    datetime_local_tz = scheduled_start_time.in_tz(pendulum.tz.local_timezone())
    scheduled_display = (
        datetime_local_tz.to_datetime_string()
        + " "
        + datetime_local_tz.tzname()
        + human_dt_diff
    )

    app.console.print(f"Created flow run {flow_run.name!r}.")
    app.console.print(
        textwrap.dedent(
            f"""
        └── UUID: {flow_run.id}
        └── Parameters: {flow_run.parameters}
        └── Job Variables: {flow_run.job_variables}
        └── Scheduled start time: {scheduled_display}
        └── URL: {run_url}
        """
        ).strip(),
        soft_wrap=True,
    )
    if watch:
        watch_interval = 5 if watch_interval is None else watch_interval
        app.console.print(f"Watching flow run {flow_run.name!r}...")
        finished_flow_run = await wait_for_flow_run(
            flow_run.id,
            timeout=watch_timeout,
            poll_interval=watch_interval,
            log_states=True,
        )
        finished_flow_run_state = finished_flow_run.state
        if finished_flow_run_state.is_completed():
            exit_with_success(
                f"Flow run finished successfully in {finished_flow_run_state.name!r}."
            )
        exit_with_error(
            f"Flow run finished in state {finished_flow_run_state.name!r}.",
            code=1,
        )

str_presenter(dumper, data)

configures yaml for dumping multiline strings Ref: https://stackoverflow.com/questions/8640959/how-can-i-control-what-scalar-form-pyyaml-uses-for-my-data

Source code in src/prefect/cli/deployment.py

def str_presenter(dumper, data):
    """
    configures yaml for dumping multiline strings
    Ref: https://stackoverflow.com/questions/8640959/how-can-i-control-what-scalar-form-pyyaml-uses-for-my-data
    """
    if len(data.splitlines()) > 1:  # check for multiline string
        return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
    return dumper.represent_scalar("tag:yaml.org,2002:str", data)