from datetime import date
from prefect import flow,task
from prefect.deployments import run_deployment
from prefect_dask.task_runners import DaskTaskRunner
from cust_test import Firstrun






@flow(task_runner=DaskTaskRunner())
@flow(name="testing")
def parent():
    deployment_name: str = "local-process"
    run_deployment(name="test/local-process",flow_run_name="parallel")
    #run_deployment.submit(flow_run_name="parallel")

if __name__ == "__main__":
    parent()

from datetime import date
from prefect import flow,task
from prefect.deployments import run_deployment
from prefect_dask.task_runners import DaskTaskRunner

@task()
def e():
    print(12)

@flow(name="test")
def Firstrun():
    e()


if __name__ == "__main__":
    Firstrun()

from my_project.flows import my_flow
from prefect.deployments import Deployment

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="example-deployment", 
    version=1, 
    work_queue_name="demo",
)
deployment.apply()

# cust_test.py

from prefect import flow,task

@task()
def e():
    print(12)

@flow(name="my_flow")
def Firstrun():
    e()


if __name__ == "__main__":
    Firstrun()

# dep.py
from cust_test import Firstrun
from prefect.deployments import Deployment, run_deployment

deployment = Deployment.build_from_flow(
    flow=Firstrun,
    name="my_deployment", 
    version=1
)
deployment.apply()
run_deployment("my_flow/my_deployment")