https://prefect.io logo
Channels
Powered by
c

Cody Webb

07/31/2023, 3:37 PM
hello all i have a flow i want to convert to a deployment and run the deployment with in python:
Copy code
@flow(name="nb-convert")
def continuous_run():
    files=["utils","hey","test"]
    [os.system(f'jupyter nbconvert --to script {file}.ipynb') for file in files]
but when i set it up for deployment:
Copy code
deployment = Deployment.build_from_flow(
        flow=continuous_run,
        name="Notebook converter poll",
        schedule=IntervalSchedule(interval=timedelta(minutes=15)),
        work_queue_name = "queue1",
    )
deployment.apply()
i get:
Copy code
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/tmp/ipykernel_102/2609488880.py in <module>
      5         work_queue_name = "queue1",
      6     )
----> 7 deployment.apply()

AttributeError: 'coroutine' object has no attribute 'apply'
Copy code
flow_run = run_deployment(
   name = "nb_converter",
   parameters = [],
   scheduled_time = IntervalSchedule(interval=timedelta(minutes=15)),
   flow_run_name = "nb-convert",
   timeout=0
)
ideally i would like to parameterize and pass in the list of files also but couldnt figure out the problem there either
d

Deceivious

07/31/2023, 3:39 PM
Are you calling the deployment.apply() in an async thread?
c

Cody Webb

07/31/2023, 3:45 PM
its in a jupyter nb currently
so yea i suppose its in an event loop
d

Deceivious

07/31/2023, 3:46 PM
check if awaiting 
Deployment.build_from_flow
fixes the deployment issues.
c

Cody Webb

07/31/2023, 3:46 PM
ok one sec lemme try
yea that kinda worked : now :
Copy code
ValueError: Could not determine flow's file location.
which im calling a local function
should i place that func in a seperate file or something?
d

Deceivious

07/31/2023, 3:47 PM
As for the paramter passing into run_deployment. You can add a parramter to the flow ex.
Copy code
@flow(name="nb-convert")
def continuous_run(ist_of_files:List):
    ....
And u can call 
run_deployment
with 
parameters={"ist_of_files":["a","b"]}
c

Cody Webb

07/31/2023, 3:47 PM
nvm got it
deployment = await Deployment.build_from_flow( flow=continuous_run(), name="Notebook converter poll", schedule=IntervalSchedule(interval=timedelta(minutes=15)), work_queue_name = "queue1", ) deployment.apply()
okay cool
alright thanks ❤️
oh ok so it seems to run but now :
Copy code
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/tmp/ipykernel_102/3257527765.py in <module>
----> 1 deployment = await Deployment.build_from_flow(
      2         flow=continuous_run(),
      3         name="Notebook converter poll",
      4         schedule=IntervalSchedule(interval=timedelta(minutes=15)),
      5         work_queue_name = "queue1",

/opt/conda/envs/rapids/lib/python3.10/site-packages/prefect/deployments/deployments.py in build_from_flow(cls, flow, name, output, skip_upload, ignore_file, apply, load_existing, **kwargs)
    790         # provided at initialization
    791         deployment = cls(name=name, **kwargs)
--> 792         deployment.flow_name = flow.name
    793         if not deployment.entrypoint:
    794             ## first see if an entrypoint can be determined

AttributeError: 'NoneType' object has no attribute 'name'
d

Deceivious

07/31/2023, 3:51 PM
flow=continuous_run(), Dont use the ()
c

Cody Webb

07/31/2023, 3:52 PM
yea but then im getting:
Copy code
ValueError: Could not determine flow's file location.
d

Deceivious

07/31/2023, 3:53 PM
build_from_flow has 
path
parameter where u can specify which py file the flow is defined in
c

Cody Webb

07/31/2023, 3:53 PM
oh so it cant take a local function? or would i just pass in the actual file this all sits in?
d

Deceivious

07/31/2023, 3:54 PM
just pass 
___file___
c

Cody Webb

07/31/2023, 3:54 PM
i see okay
thx
okay so i got the deployment built with:
Copy code
from nb_converter import continuous_run
deployment = await Deployment.build_from_flow(
        flow=continuous_run,
        name="Notebook converter poll",
        #schedule=IntervalSchedule(interval=timedelta(minutes=15)),
        work_queue_name = "queue_1",
    )
deploy_id = await deployment.apply()
and im going to run it with
Copy code
flow_run = await run_deployment(
   name = deploy_id,
   scheduled_time = IntervalSchedule(interval=timedelta(minutes=15)),
   flow_run_name = "Notebook converter poll",
   timeout=0
)
await flow_run.apply()
but im getting:
Copy code
PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://127.0.0.1:4200/api/deployments/688d6991-6ede-4a36-859b-218208bae56e/create_flow_run>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['body', 'state', 'state_details', 'scheduled_time'], 'msg': 'Unrecognized datetime.', 'type': 'value_error'}], 'request_body': {'state': {'type': 'SCHEDULED', 'name': 'Scheduled', 'message': None, 'state_details': {'scheduled_time': {'interval': 900.0}}, 'data': None}, 'name': 'busy-cow', 'parameters': {}, 'context': {}, 'tags': [], 'idempotency_key': None, 'parent_task_run_id': None}}
For more information check: <https://httpstatuses.com/422>
can i not pass in a interval schedule here for scheduled_time?
nvm ill just set schedule in build_from_flow and do a launch now for run_deployment..thanks! been a big help
Powered by