AttributeError: 'NoneType' object has no attribute 'call' on sync_portal

yesparkpe · September 21, 2022, 6:21pm

Hello,
I’m at a loss to explain why this flow fails and succeeds at the same time… It only consists in triggering a dbt build, which succeeds, but somehow the flow itself fails.

Here’s the error:

The error message:

    return flow_run_context.sync_portal.call(begin_run)
AttributeError: 'NoneType' object has no attribute 'call'

And the code:

@flow(name='DBT Incremental build')
def trigger_dbt_build():
    run_result = trigger_dbt_cloud_job_run_and_wait_for_completion(
        dbt_cloud_credentials=DbtCloudCredentials(
            api_key=Secret.load("dbt-api-key").get(),
            account_id=int(Secret.load("dbt-account-id").get())
        ),
        job_id=DBT_UPDATE_PROD_INCREMENTAL_JOB_ID,
        max_wait_seconds=1200
    )
    return run_result

@flow
def run_pipeline():
    run_fivetran()
    trigger_dbt_build()
    sync_metabase()

Thanks for your help

anna_geller · September 22, 2022, 2:23pm

I think the issue might be with returning something

Can you try this?

@flow(name='DBT Incremental build')
def trigger_dbt_build():
    trigger_dbt_cloud_job_run_and_wait_for_completion(
        dbt_cloud_credentials=DbtCloudCredentials(
            api_key=Secret.load("dbt-api-key").get(),
            account_id=int(Secret.load("dbt-account-id").get())
        ),
        job_id=DBT_UPDATE_PROD_INCREMENTAL_JOB_ID,
        max_wait_seconds=1200
    )

@flow
def run_pipeline():
    run_fivetran()
    trigger_dbt_build()
    sync_metabase()

btw for DbtCloudCredentials, you should be able to set this up once in the UI e.g. with a block name “prod”, and then only load it in your flow like so:

from prefect_dbt.cloud.credentials import DbtCloudCredentials


@flow(name='DBT Incremental build')
def trigger_dbt_build():
    trigger_dbt_cloud_job_run_and_wait_for_completion(
        dbt_cloud_credentials=DbtCloudCredentials.load("prod"),
        job_id=DBT_UPDATE_PROD_INCREMENTAL_JOB_ID,
        max_wait_seconds=1200
    )

yesparkpe · September 22, 2022, 3:00pm

Thanks for your reply! I actually had also tried to include the trigger_dbt_cloud_job_run_and_wait_for_completion in run_pipeline(), to no avail. Just tried again to make sure, I’ll keep you posted!
As a side note, would you be able to explain how returning something might be an issue?

yesparkpe · September 22, 2022, 3:22pm

So I inlined trigger_dbt_cloud_job_run_and_wait_for_completion in the main method as well as hardcoded the credentials, but it still fails with the same error…

anna_geller · September 22, 2022, 3:36pm

Can you share your prefect version output?

anna_geller · September 22, 2022, 3:36pm

Also what prefect-dbt version are you currently using? Can yu try upgrading both packages in a new Conda environment?

yesparkpe · September 22, 2022, 3:54pm

Both versions are the latest, running in a Docker container on a remote server

# docker exec -it root-agent-1 prefect --version
2.4.0
# docker exec -it root-agent-1 pip list | grep dbt
prefect-dbt        0.2.2

yesparkpe · September 22, 2022, 4:07pm

Here’s the whole stacktrace from the main flow

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 595, in orchestrate_flow_run
    result = await run_sync(flow_call)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 57, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/tmp/tmpovddxriwprefect/flows/yespark_pipeline.py", line 31, in run_pipeline
    trigger_dbt_cloud_job_run_and_wait_for_completion(
  File "/usr/local/lib/python3.10/site-packages/prefect/flows.py", line 384, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 162, in enter_flow_run_engine_from_flow_call
    return run_async_from_worker_thread(begin_run)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 137, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/usr/local/lib/python3.10/site-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 458, in result
    return self.__get_result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/site-packages/prefect/client.py", line 103, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 521, in create_and_begin_subflow_run
    return terminal_state.result()
  File "/usr/local/lib/python3.10/site-packages/prefect/orion/schemas/states.py", line 143, in result
    raise data
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 588, in orchestrate_flow_run
    result = await flow_call()
  File "/usr/local/lib/python3.10/site-packages/prefect_dbt/cloud/jobs.py", line 287, in trigger_dbt_cloud_job_run_and_wait_for_completion
    run_id_future = get_run_id.submit(triggered_run_data_future)
  File "/usr/local/lib/python3.10/site-packages/prefect/tasks.py", line 492, in submit
    return enter_task_run_engine(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 740, in enter_task_run_engine
    return flow_run_context.sync_portal.call(begin_run)
AttributeError: 'NoneType' object has no attribute 'call'

anna_geller · September 22, 2022, 4:10pm

Could you share the full flow code? it looks like the part that you shared works, but only the one using Fivetran doesn’t.

yesparkpe · September 22, 2022, 4:16pm

Actually I might not have been clear enough, but launching the dbt flow independently, without a return as you suggested earlier, fails in the same way

anna_geller · September 22, 2022, 4:37pm

Can you try the same using the free version of Cloud: https://app.prefect.cloud/? It would be interesting to see if the issue is in the orchestration layer or execution layer

yesparkpe · September 23, 2022, 10:29am

Good afternoon Anna,
I tried running everything locally (as opposed to Prefect Cloud UI <-> self-hosted remote agent), still fails with the same error

yesparkpe · September 23, 2022, 11:10am

Apparently there’s an open issue on prefect-dbt referring a Prefect PR Fixes sync calling async subflows, not sure how I missed it

I rewrote the method like so (after trying various combinations with/without return with/without await:

@flow(name='DBT Incremental build')
async def trigger_dbt_build():
    dbt_api_key_secret = await Secret.load("dbt-api-key")
    dbt_api_key = dbt_api_key_secret.get()
    dbt_account_id_secret = await Secret.load("dbt-account-id")
    dbt_account_id = int(dbt_account_id_secret.get())

    return await trigger_dbt_cloud_job_run_and_wait_for_completion(
        dbt_cloud_credentials=DbtCloudCredentials(
            api_key=dbt_api_key,
            account_id=dbt_account_id
        ),
        job_id=DBT_UPDATE_PROD_INCREMENTAL_JOB_ID,
        max_wait_seconds=1200
    )

and now the dbt job is still created and runs normally, but the flow fails right afterwards with the following error:

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 589, in orchestrate_flow_run
    result = await flow_call()
  File "/usr/local/lib/python3.10/site-packages/prefect_dbt/cloud/jobs.py", line 289, in trigger_dbt_cloud_job_run_and_wait_for_completion
    final_run_status, run_data = await wait_for_dbt_cloud_job_run(
  File "/usr/local/lib/python3.10/site-packages/prefect/client/orion.py", line 80, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 429, in create_and_begin_subflow_run
    parameters = await resolve_inputs(parameters)
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1407, in resolve_inputs
    return await run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 57, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/collections.py", line 307, in visit_collection
    items = [(visit_nested(k), visit_nested(v)) for k, v in expr.items()]
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/collections.py", line 307, in <listcomp>
    items = [(visit_nested(k), visit_nested(v)) for k, v in expr.items()]
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/collections.py", line 273, in visit_nested
    return visit_collection(
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/collections.py", line 281, in visit_collection
    result = visit_fn(expr)
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1400, in resolve_input
    raise UpstreamTaskError(
prefect.exceptions.UpstreamTaskError: Upstream task run 'None' did not reach a 'COMPLETED' state.
01:07:28 PM
Crash detected! Execution was interrupted by an unexpected exception.

anna_geller · September 23, 2022, 11:21am

Interesting, thanks so much for digging deeper. Just curious, if you don’t poll for the job status and don’t wait for its completion, does it work then?

bogdan.cioata · September 23, 2022, 11:29am

@anna_geller: just triggering the job works completely fine.

yesparkpe · September 30, 2022, 7:56am

@bogdan.cioata Thanks for chiming in! So to date you’re still having this issue? Just upgraded to 2.4.5 and it’s still occurring on every run…

anna_geller · September 30, 2022, 11:23am

I’d encourage you to chime in on the GitHub issue on the prefect-dbt repo dbt flow fails to sync back · Issue #67 · PrefectHQ/prefect-dbt · GitHub - this will make it easier to follow once this is fixed

yesparkpe · October 5, 2022, 9:02am

FIXED!

anna_geller · October 5, 2022, 6:29pm

thanks for reporting back

Topic		Replies	Views
Bug when running sub flow it hangs in pending and main flow errors with AttributeError: 'Flow' object has no attribute 'result_storage' Help	1	12	July 19, 2024
Flow run crashes with AttributeError: 'Flow' object has no attribute 'should_validate_parameters' Help	2	1985	May 12, 2023
When using the S3Upload task from the task library, I'm getting an error: "AttributeError: 'str' object has no attribute 'bucket'" - how to fix this? Archive prefect-1-0 , dag-flow-structure , aws , s3 , imperative-api , task-library	0	785	February 14, 2022
Error when using Prefect 2.0: AttributeError: type object 'ORMInsert' has no attribute 'orm_pre_session_exec' Archive prefect-2-0	5	1562	April 1, 2022
AttributeError: 'Flow' object has no attribute 'serve' Help	1	220	April 30, 2024

AttributeError: 'NoneType' object has no attribute 'call' on sync_portal

Related topics