-
Notifications
You must be signed in to change notification settings - Fork 0
/
deploy_flow.py
57 lines (49 loc) · 1.8 KB
/
deploy_flow.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
"""
This file is used to set up the Prefect deployment for invoking lambda functions.
"""
from typing import Optional
from common_lambda_trigger.invoke import invoke_lambda_function
from prefect.deployments import Deployment
from prefect.filesystems import S3
from prefect.server.schemas.schedules import CronSchedule
def deploy(
lambda_function_name: str,
cron_schedule: str,
tags: Optional[list] = None,
version: Optional[str] = None,
invocation_type: Optional[str] = None,
payload: Optional[dict] = None,
):
"""Deploy the flow to Prefect Cloud."""
try:
# Load S3 Storage for flow deployment:
s3_storage = S3.load("s3-deployments-block")
# Build flow parameters to pass to the deployment:
deployment_parameters = {
"function_name": lambda_function_name,
}
if invocation_type:
deployment_parameters["invocation_type"] = invocation_type
if payload:
deployment_parameters["payload"] = payload
# Build the deployment configuration:
deployment = Deployment.build_from_flow(
flow=invoke_lambda_function,
flow_name=f"{lambda_function_name}-flow",
name=f"{lambda_function_name}-deployment",
tags=tags if tags else ["common", "railway.app", "deployment"],
version=version if version else "v1.0.0",
storage=s3_storage,
schedule=CronSchedule(cron=cron_schedule, timezone="America/New_York"),
is_schedule_active=True,
parameters=deployment_parameters,
)
deployment.apply()
print("Deployment successful!")
except Exception as e:
raise e
if __name__ == "__main__":
deploy(
lambda_function_name="hello-world",
cron_schedule="0 8 * * *",
)