54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165 | @flow_app.command()
async def serve(
entrypoint: str = typer.Argument(
...,
help=(
"The path to a file containing a flow and the name of the flow function in"
" the format `./path/to/file.py:flow_func_name`."
),
),
name: str = typer.Option(
...,
"--name",
"-n",
help="The name to give the deployment created for the flow.",
),
description: Optional[str] = typer.Option(
None,
"--description",
"-d",
help=(
"The description to give the created deployment. If not provided, the"
" description will be populated from the flow's description."
),
),
version: Optional[str] = typer.Option(
None, "-v", "--version", help="A version to give the created deployment."
),
tags: Optional[List[str]] = typer.Option(
None,
"-t",
"--tag",
help="One or more optional tags to apply to the created deployment.",
),
cron: Optional[str] = typer.Option(
None,
"--cron",
help=(
"A cron string that will be used to set a schedule for the created"
" deployment."
),
),
interval: Optional[int] = typer.Option(
None,
"--interval",
help=(
"An integer specifying an interval (in seconds) between scheduled runs of"
" the flow."
),
),
interval_anchor: Optional[str] = typer.Option(
None, "--anchor-date", help="The start date for an interval schedule."
),
rrule: Optional[str] = typer.Option(
None,
"--rrule",
help="An RRule that will be used to set a schedule for the created deployment.",
),
timezone: Optional[str] = typer.Option(
None,
"--timezone",
help="Timezone to used scheduling flow runs e.g. 'America/New_York'",
),
pause_on_shutdown: bool = typer.Option(
True,
help=(
"If set, provided schedule will be paused when the serve command is"
" stopped. If not set, the schedules will continue running."
),
),
):
"""
Serve a flow via an entrypoint.
"""
runner = Runner(name=name, pause_on_shutdown=pause_on_shutdown)
try:
schedules = []
if interval or cron or rrule:
schedule = construct_schedule(
interval=interval,
cron=cron,
rrule=rrule,
timezone=timezone,
anchor_date=interval_anchor,
)
schedules = [MinimalDeploymentSchedule(schedule=schedule, active=True)]
runner_deployment = RunnerDeployment.from_entrypoint(
entrypoint=entrypoint,
name=name,
schedules=schedules,
description=description,
tags=tags or [],
version=version,
)
except (MissingFlowError, ValueError) as exc:
exit_with_error(str(exc))
deployment_id = await runner.add_deployment(runner_deployment)
help_message = (
f"[green]Your flow {runner_deployment.flow_name!r} is being served and polling"
" for scheduled runs!\n[/]\nTo trigger a run for this flow, use the following"
" command:\n[blue]\n\t$ prefect deployment run"
f" '{runner_deployment.flow_name}/{name}'\n[/]"
)
if PREFECT_UI_URL:
help_message += (
"\nYou can also run your flow via the Prefect UI:"
f" [blue]{PREFECT_UI_URL.value()}/deployments/deployment/{deployment_id}[/]\n"
)
app.console.print(help_message, soft_wrap=True)
await runner.start()
|