register_flow.py

import prefect
from prefect import task, Flow, Parameter
from prefect.utilities.storage import extract_flow_from_file
from prefect.run_configs import DockerRun
import datetime as dt
import argparse

@task(checkpoint=False,log_stdout=True,max_retries=3,retry_delay=dt.timedelta(seconds=10))
def register_flow(path, project_name):
    flow = extract_flow_from_file(path)
    return flow.register(project_name=project_name)

with Flow("Register Flow") as flow:
    flow_path = Parameter("flow_path")
    project_name = Parameter("project_name")

    flow_id = register_flow(flow_path, project_name)

# flow.run_config = DockerRun(image="prefect-hello-world-dockerized:latest")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Register Flow')
    parser.add_argument('--project_name', dest='project_name', required=True)
    parser.add_argument('--flow_path', dest='flow_path', required=True)
    parser.add_argument('--image_name', dest='image_name', required=True)
    args = parser.parse_args()
    flow.run_config = DockerRun(image=args.image_name)

    flow.run(parameters={
        'project_name': args.project_name,
        'flow_path': args.flow_path
    })