n

    Noam polak

    10 months ago
    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 = stringinput_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

    Kevin Kho

    10 months ago
    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?
    n

    Noam polak

    10 months ago
    portfolio_uuid =  stringinput_row_size = int
    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

    Kevin Kho

    10 months ago
    This error makes it seem like it’s a different type. Can you log the types maybe in a task before start_flow_run?
    @task(log_stdout=True)
    def get_types(portfolio_uuid, input_row_size):
        print(type(portfolio_uuid))
        print(type(input_row_size))
        return
    n

    Noam polak

    10 months ago
    yes
    portfolio_uuid =<class 'str'> input_row_size = <class 'method'>
    Weird
    I think I found the problem