Hey everyone slightly smiling face I have trouble with trigg Prefect Community #ask-community

Hey everyone :slightly_smiling_face: I have troubl...

Noam polak

11/09/2021, 3:52 PM

Hey everyone 🙂 I have trouble with triggering flow inside flow I have a flow with (I'm shortening):

with Flow(

flow_name,

run_config=run_config_provider(flow_name),

result=results_provider(os.environ["GCP_PROJECT"]),

storage=storage_provider(flow_name),

) as flow:

portfolio = fetch_portfolio(portfolio_uuid)

parser_results, results_log, csv_log, *input_row_size* = run_parser(

portfolio, excel_bucket_url, workflow_id

.... more tasks...

# run the next flow

with case(parser_results, "completed"):

start_flow_run = StartFlowRun(

flow_name=automation_flow_name,

project_name="project_name",

start_flow_run(

upstream_tasks=[updated, *input_row_size*],

parameters={

"portfolio_uuid": portfolio_uuid,

"input_row_size": *input_row_size*,

},

portfolio_uuid = string input_row_size = int When I run the flow it fails when attempting to run the new flow here is the error message (in the message thread): What did I do that was wrong?

Kevin Kho

11/09/2021, 3:54 PM

Hey @Noam polak, could we move the traceback to the thread to keep the main channel a bit more compact? When you start a flow run, this is hitting the Prefect API under the hood. Parameters that get passed through APIs need to be JSONSerializeable. What are the types of your parameters here?

Noam polak

11/09/2021, 3:55 PM

portfolio_uuid = string input_row_size = int

Noam polak

11/09/2021, 3:56 PM

Task 'Flow automation': Exception encountered during task execution! Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 861, in get_task_run_state value = prefect.utilities.executors.run_task_with_timeout( File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 327, in run_task_with_timeout return task.run(*args, **kwargs) # type: ignore File "/usr/local/lib/python3.8/site-packages/prefect/utilities/tasks.py", line 441, in method return run_method(self, *args, **kwargs) File "/usr/local/lib/python3.8/site-packages/prefect/tasks/prefect/flow_run.py", line 418, in run flow_run_id = client.create_flow_run( File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1412, in create_flow_run res = self.graphql(create_mutation, variables=dict(input=inputs)) File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 532, in graphql params=dict(query=parse_graphql(query), variables=json.dumps(variables)), File "/usr/local/lib/python3.8/json/__init__.py", line 231, in dumps return _default_encoder.encode(obj) File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode chunks = self.iterencode(o, _one_shot=True) File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode return _iterencode(o, 0) File "/usr/local/lib/python3.8/json/encoder.py", line 179, in default raise TypeError(f'Object of type {o.class.name} ' TypeError: Object of type method is not JSON serializable

Kevin Kho

11/09/2021, 4:01 PM

This error makes it seem like it’s a different type. Can you log the types maybe in a task before start_flow_run?

Copy code

@task(log_stdout=True)
def get_types(portfolio_uuid, input_row_size):
    print(type(portfolio_uuid))
    print(type(input_row_size))
    return

Noam polak

11/09/2021, 4:03 PM

yes

Noam polak

11/09/2021, 4:25 PM

portfolio_uuid =<class 'str'> input_row_size = <class 'method'>

Noam polak

11/09/2021, 4:25 PM

Weird

Noam polak

11/09/2021, 4:29 PM

I think I found the problem

Noam polak

11/09/2021, 4:29 PM

thanks

👍 1

7 Views

Open in Slack

Previous Next