prefect.cli.deployment

Command line interface for working with deployments.

apply async

Create or update a deployment from a YAML file.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command()
async def apply(
    paths: List[str] = typer.Argument(
        ...,
        help="One or more paths to deployment YAML files.",
    ),
    upload: bool = typer.Option(
        False,
        "--upload",
        help=(
            "A flag that, when provided, uploads this deployment's files to remote"
            " storage."
        ),
    ),
    work_queue_concurrency: int = typer.Option(
        None,
        "--limit",
        "-l",
        help=(
            "Sets the concurrency limit on the work queue that handles this"
            " deployment's runs"
        ),
    ),
):
    """
    Create or update a deployment from a YAML file.
    """
    async with get_client() as client:
        for path in paths:
            try:
                deployment = await Deployment.load_from_yaml(path)
                app.console.print(
                    f"Successfully loaded {deployment.name!r}", style="green"
                )
            except Exception as exc:
                exit_with_error(
                    f"'{path!s}' did not conform to deployment spec: {exc!r}"
                )

            await create_work_queue_and_set_concurrency_limit(
                deployment.work_queue_name,
                deployment.work_pool_name,
                work_queue_concurrency,
            )

            if upload:
                if (
                    deployment.storage
                    and "put-directory" in deployment.storage.get_block_capabilities()
                ):
                    file_count = await deployment.upload_to_storage()
                    if file_count:
                        app.console.print(
                            (
                                f"Successfully uploaded {file_count} files to"
                                f" {deployment.location}"
                            ),
                            style="green",
                        )
                else:
                    app.console.print(
                        (
                            f"Deployment storage {deployment.storage} does not have"
                            " upload capabilities; no files uploaded."
                        ),
                        style="red",
                    )
            await check_work_pool_exists(
                work_pool_name=deployment.work_pool_name, client=client
            )
            deployment_id = await deployment.apply()
            app.console.print(
                (
                    f"Deployment '{deployment.flow_name}/{deployment.name}'"
                    f" successfully created with id '{deployment_id}'."
                ),
                style="green",
            )

            if PREFECT_UI_URL:
                app.console.print(
                    "View Deployment in UI:"
                    f" {PREFECT_UI_URL.value()}/deployments/deployment/{deployment_id}"
                )

            if deployment.work_pool_name is not None:
                await _print_deployment_work_pool_instructions(
                    work_pool_name=deployment.work_pool_name, client=client
                )
            elif deployment.work_queue_name is not None:
                app.console.print(
                    "\nTo execute flow runs from this deployment, start an agent that"
                    f" pulls work from the {deployment.work_queue_name!r} work queue:"
                )
                app.console.print(
                    f"$ prefect agent start -q {deployment.work_queue_name!r}",
                    style="blue",
                )
            else:
                app.console.print(
                    (
                        "\nThis deployment does not specify a work queue name, which"
                        " means agents will not be able to pick up its runs. To add a"
                        " work queue, edit the deployment spec and re-run this command,"
                        " or visit the deployment in the UI."
                    ),
                    style="red",
                )

build async

Generate a deployment YAML from /path/to/file.py:flow_function

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command()
async def build(
    entrypoint: str = typer.Argument(
        ...,
        help=(
            "The path to a flow entrypoint, in the form of"
            " `./path/to/file.py:flow_func_name`"
        ),
    ),
    name: str = typer.Option(
        None, "--name", "-n", help="The name to give the deployment."
    ),
    description: str = typer.Option(
        None,
        "--description",
        "-d",
        help=(
            "The description to give the deployment. If not provided, the description"
            " will be populated from the flow's description."
        ),
    ),
    version: str = typer.Option(
        None, "--version", "-v", help="A version to give the deployment."
    ),
    tags: List[str] = typer.Option(
        None,
        "-t",
        "--tag",
        help=(
            "One or more optional tags to apply to the deployment. Note: tags are used"
            " only for organizational purposes. For delegating work to agents, use the"
            " --work-queue flag."
        ),
    ),
    work_queue_name: str = typer.Option(
        None,
        "-q",
        "--work-queue",
        help=(
            "The work queue that will handle this deployment's runs. "
            "It will be created if it doesn't already exist. Defaults to `None`. "
            "Note that if a work queue is not set, work will not be scheduled."
        ),
    ),
    work_pool_name: str = typer.Option(
        None,
        "-p",
        "--pool",
        help="The work pool that will handle this deployment's runs.",
    ),
    work_queue_concurrency: int = typer.Option(
        None,
        "--limit",
        "-l",
        help=(
            "Sets the concurrency limit on the work queue that handles this"
            " deployment's runs"
        ),
    ),
    infra_type: str = typer.Option(
        None,
        "--infra",
        "-i",
        help="The infrastructure type to use, prepopulated with defaults. For example: "
        + listrepr(builtin_infrastructure_types, sep=", "),
    ),
    infra_block: str = typer.Option(
        None,
        "--infra-block",
        "-ib",
        help="The slug of the infrastructure block to use as a template.",
    ),
    overrides: List[str] = typer.Option(
        None,
        "--override",
        help=(
            "One or more optional infrastructure overrides provided as a dot delimited"
            " path, e.g., `env.env_key=env_value`"
        ),
    ),
    storage_block: str = typer.Option(
        None,
        "--storage-block",
        "-sb",
        help=(
            "The slug of a remote storage block. Use the syntax:"
            " 'block_type/block_name', where block_type is one of 'github', 's3',"
            " 'gcs', 'azure', 'smb', or a registered block from a library that"
            " implements the WritableDeploymentStorage interface such as"
            " 'gitlab-repository', 'bitbucket-repository', 's3-bucket',"
            " 'gcs-bucket'"
        ),
    ),
    skip_upload: bool = typer.Option(
        False,
        "--skip-upload",
        help=(
            "A flag that, when provided, skips uploading this deployment's files to"
            " remote storage."
        ),
    ),
    cron: str = typer.Option(
        None,
        "--cron",
        help="A cron string that will be used to set a CronSchedule on the deployment.",
    ),
    interval: int = typer.Option(
        None,
        "--interval",
        help=(
            "An integer specifying an interval (in seconds) that will be used to set an"
            " IntervalSchedule on the deployment."
        ),
    ),
    interval_anchor: Optional[str] = typer.Option(
        None, "--anchor-date", help="The anchor date for an interval schedule"
    ),
    rrule: str = typer.Option(
        None,
        "--rrule",
        help="An RRule that will be used to set an RRuleSchedule on the deployment.",
    ),
    timezone: str = typer.Option(
        None,
        "--timezone",
        help="Deployment schedule timezone string e.g. 'America/New_York'",
    ),
    path: str = typer.Option(
        None,
        "--path",
        help=(
            "An optional path to specify a subdirectory of remote storage to upload to,"
            " or to point to a subdirectory of a locally stored flow."
        ),
    ),
    output: str = typer.Option(
        None,
        "--output",
        "-o",
        help="An optional filename to write the deployment file to.",
    ),
    _apply: bool = typer.Option(
        False,
        "--apply",
        "-a",
        help=(
            "An optional flag to automatically register the resulting deployment with"
            " the API."
        ),
    ),
    param: List[str] = typer.Option(
        None,
        "--param",
        help=(
            "An optional parameter override, values are parsed as JSON strings e.g."
            " --param question=ultimate --param answer=42"
        ),
    ),
    params: str = typer.Option(
        None,
        "--params",
        help=(
            "An optional parameter override in a JSON string format e.g."
            ' --params=\'{"question": "ultimate", "answer": 42}\''
        ),
    ),
):
    """
    Generate a deployment YAML from /path/to/file.py:flow_function
    """
    # validate inputs
    if not name:
        exit_with_error(
            "A name for this deployment must be provided with the '--name' flag."
        )

    if len([value for value in (cron, rrule, interval) if value is not None]) > 1:
        exit_with_error("Only one schedule type can be provided.")

    if infra_block and infra_type:
        exit_with_error(
            "Only one of `infra` or `infra_block` can be provided, please choose one."
        )

    output_file = None
    if output:
        output_file = Path(output)
        if output_file.suffix and output_file.suffix != ".yaml":
            exit_with_error("Output file must be a '.yaml' file.")
        else:
            output_file = output_file.with_suffix(".yaml")

    # validate flow
    try:
        fpath, obj_name = entrypoint.rsplit(":", 1)
    except ValueError as exc:
        if str(exc) == "not enough values to unpack (expected 2, got 1)":
            missing_flow_name_msg = (
                "Your flow entrypoint must include the name of the function that is"
                f" the entrypoint to your flow.\nTry {entrypoint}:<flow_name>"
            )
            exit_with_error(missing_flow_name_msg)
        else:
            raise exc
    try:
        flow = await run_sync_in_worker_thread(load_flow_from_entrypoint, entrypoint)
    except Exception as exc:
        exit_with_error(exc)
    app.console.print(f"Found flow {flow.name!r}", style="green")
    infra_overrides = {}
    for override in overrides or []:
        key, value = override.split("=", 1)
        infra_overrides[key] = value

    if infra_block:
        infrastructure = await Block.load(infra_block)
    elif infra_type:
        # Create an instance of the given type
        infrastructure = Block.get_block_class_from_key(infra_type)()
    else:
        # will reset to a default of Process is no infra is present on the
        # server-side definition of this deployment
        infrastructure = None

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

    schedule = None
    if cron:
        cron_kwargs = {"cron": cron, "timezone": timezone}
        schedule = CronSchedule(
            **{k: v for k, v in cron_kwargs.items() if v is not None}
        )
    elif interval:
        interval_kwargs = {
            "interval": timedelta(seconds=interval),
            "anchor_date": interval_anchor,
            "timezone": timezone,
        }
        schedule = IntervalSchedule(
            **{k: v for k, v in interval_kwargs.items() if v is not None}
        )
    elif rrule:
        try:
            schedule = RRuleSchedule(**json.loads(rrule))
            if timezone:
                # override timezone if specified via CLI argument
                schedule.timezone = timezone
        except json.JSONDecodeError:
            schedule = RRuleSchedule(rrule=rrule, timezone=timezone)

    # parse storage_block
    if storage_block:
        block_type, block_name, *block_path = storage_block.split("/")
        if block_path and path:
            exit_with_error(
                "Must provide a `path` explicitly or provide one on the storage block"
                " specification, but not both."
            )
        elif not path:
            path = "/".join(block_path)
        storage_block = f"{block_type}/{block_name}"
        storage = await Block.load(storage_block)
    else:
        storage = None

    if create_default_ignore_file(path="."):
        app.console.print(
            (
                "Default '.prefectignore' file written to"
                f" {(Path('.') / '.prefectignore').absolute()}"
            ),
            style="green",
        )

    if param and (params is not None):
        exit_with_error("Can only pass one of `param` or `params` options")

    parameters = dict()

    if param:
        for p in param or []:
            k, unparsed_value = p.split("=", 1)
            try:
                v = json.loads(unparsed_value)
                app.console.print(
                    f"The parameter value {unparsed_value} is parsed as a JSON string"
                )
            except json.JSONDecodeError:
                v = unparsed_value
            parameters[k] = v

    if params is not None:
        parameters = json.loads(params)

    # set up deployment object
    entrypoint = (
        f"{Path(fpath).absolute().relative_to(Path('.').absolute())}:{obj_name}"
    )

    init_kwargs = dict(
        path=path,
        entrypoint=entrypoint,
        version=version,
        storage=storage,
        infra_overrides=infra_overrides or {},
    )

    if parameters:
        init_kwargs["parameters"] = parameters

    if description:
        init_kwargs["description"] = description

    # if a schedule, tags, work_queue_name, or infrastructure are not provided via CLI,
    # we let `build_from_flow` load them from the server
    if schedule:
        init_kwargs.update(schedule=schedule)
    if tags:
        init_kwargs.update(tags=tags)
    if infrastructure:
        init_kwargs.update(infrastructure=infrastructure)
    if work_queue_name:
        init_kwargs.update(work_queue_name=work_queue_name)
    if work_pool_name:
        init_kwargs.update(work_pool_name=work_pool_name)

    deployment_loc = output_file or f"{obj_name}-deployment.yaml"
    deployment = await Deployment.build_from_flow(
        flow=flow,
        name=name,
        output=deployment_loc,
        skip_upload=skip_upload,
        apply=False,
        **init_kwargs,
    )
    app.console.print(
        f"Deployment YAML created at '{Path(deployment_loc).absolute()!s}'.",
        style="green",
    )

    await create_work_queue_and_set_concurrency_limit(
        deployment.work_queue_name, deployment.work_pool_name, work_queue_concurrency
    )

    # we process these separately for informative output
    if not skip_upload:
        if (
            deployment.storage
            and "put-directory" in deployment.storage.get_block_capabilities()
        ):
            file_count = await deployment.upload_to_storage()
            if file_count:
                app.console.print(
                    (
                        f"Successfully uploaded {file_count} files to"
                        f" {deployment.location}"
                    ),
                    style="green",
                )
        else:
            app.console.print(
                (
                    f"Deployment storage {deployment.storage} does not have upload"
                    " capabilities; no files uploaded.  Pass --skip-upload to suppress"
                    " this warning."
                ),
                style="green",
            )

    if _apply:
        async with get_client() as client:
            await check_work_pool_exists(
                work_pool_name=deployment.work_pool_name, client=client
            )
            deployment_id = await deployment.apply()
            app.console.print(
                (
                    f"Deployment '{deployment.flow_name}/{deployment.name}'"
                    f" successfully created with id '{deployment_id}'."
                ),
                style="green",
            )
            if deployment.work_pool_name is not None:
                await _print_deployment_work_pool_instructions(
                    work_pool_name=deployment.work_pool_name, client=client
                )

            elif deployment.work_queue_name is not None:
                app.console.print(
                    "\nTo execute flow runs from this deployment, start an agent that"
                    f" pulls work from the {deployment.work_queue_name!r} work queue:"
                )
                app.console.print(
                    f"$ prefect agent start -q {deployment.work_queue_name!r}",
                    style="blue",
                )
            else:
                app.console.print(
                    (
                        "\nThis deployment does not specify a work queue name, which"
                        " means agents will not be able to pick up its runs. To add a"
                        " work queue, edit the deployment spec and re-run this command,"
                        " or visit the deployment in the UI."
                    ),
                    style="red",
                )

delete async

Delete a deployment.



Examples:

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

Source code in /home/runner/work/docs/docs/prefect_source/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:
                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)
                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")

inspect 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', 'schedule': None, 'is_schedule_active': True, '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 /home/runner/work/docs/docs/prefect_source/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',
            'schedule': None,
            'is_schedule_active': True,
            '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.dict(json_compatible=True)

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

    app.console.print(Pretty(deployment_json))

ls async

View all deployments or deployments for specific flows.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command()
async def ls(flow_name: 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 async

Pause schedule of a given deployment.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command("pause-schedule")
async def pause_schedule(
    name: str,
):
    """
    Pause schedule of a given deployment.
    """
    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!")

        await client.update_deployment(deployment, is_schedule_active=False)
        exit_with_success(f"Paused schedule for deployment {name}")

resume_schedule async

Resume schedule of a given deployment.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command("resume-schedule")
async def resume_schedule(
    name: str,
):
    """
    Resume schedule of a given deployment.
    """
    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!")

        await client.update_deployment(deployment, is_schedule_active=True)
        exit_with_success(f"Resumed schedule for deployment {name}")

run 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 an agent starts.

Source code in /home/runner/work/docs/docs/prefect_source/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"
    ),
    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",
    ),
    start_at: Optional[str] = typer.Option(
        None,
        "--start-at",
    ),
):
    """
    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 an agent starts.
    """
    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}")

    cli_params = _load_json_key_values(params, "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}

    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()),
                    },
                )

            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}'...",
        )

        flow_run = await client.create_flow_run_from_deployment(
            deployment.id,
            parameters=parameters,
            state=Scheduled(scheduled_time=scheduled_start_time),
        )

    if PREFECT_UI_URL:
        run_url = f"{PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
    else:
        run_url = "<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}
        └── Scheduled start time: {scheduled_display}
        └── URL: {run_url}
        """
        ).strip()
    )

set_schedule async

Set schedule for a given deployment.

Source code in /home/runner/work/docs/docs/prefect_source/src/prefect/cli/deployment.py

@deployment_app.command("set-schedule")
async def set_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'",
    ),
):
    """
    Set 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."
        )

    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:
                exit_with_error("The anchor date must be a valid date string.")
        interval_schedule = {
            "interval": interval,
            "anchor_date": interval_anchor,
            "timezone": timezone,
        }
        updated_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,
        }
        updated_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:
            updated_schedule = RRuleSchedule(**json.loads(rrule_string))
            if timezone:
                # override timezone if specified via CLI argument
                updated_schedule.timezone = timezone
        except json.JSONDecodeError:
            updated_schedule = RRuleSchedule(rrule=rrule_string, timezone=timezone)

    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!")

        await client.update_deployment(deployment, schedule=updated_schedule)
        exit_with_success("Updated deployment schedule!")

str_presenter

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 /home/runner/work/docs/docs/prefect_source/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)