Also - anyone have any idea why I'm getting the following lengthy error when I manually run a flow from the UI? The flow runs fine locally, the agent's running, it's a simple DBT project. I followed the docs exactly. Thoughts?

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 883, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "C:\Users\36350admin\AppData\Local\Temp\2\tmp6kmb0zxkprefect\dbt_flow.py", line 35, in <module>
    run_dbt()
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\flows.py", line 388, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\engine.py", line 159, in enter_flow_run_engine_from_flow_call
    return anyio.run(begin_run)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\_core\_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\_backends\_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "C:\Python\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Python\lib\asyncio\base_events.py", line 646, in run_until_complete
    return future.result()
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\_backends\_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\client\orion.py", line 82, in with_injected_client
    return await fn(*args, **kwargs)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\engine.py", line 232, in create_then_begin_flow_run
    state = await begin_flow_run(
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\engine.py", line 365, in begin_flow_run
    await result_filesystem._save(is_anonymous=True)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\client\orion.py", line 82, in with_injected_client
    return await fn(*args, **kwargs)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\blocks\core.py", line 760, in _save
    await self.register_type_and_schema(client=client)
TypeError: object NoneType can't be used in 'await' expression

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\engine.py", line 257, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\client\orion.py", line 82, in with_injected_client
    return await fn(*args, **kwargs)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\deployments.py", line 70, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(import_object, str(import_path))
  File "C:\Users\36350admin\prefect\venv\lib\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 "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\_backends\_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\anyio\_backends\_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\utilities\importtools.py", line 193, in import_object
    module = load_script_as_module(script_path)
  File "C:\Users\36350admin\prefect\venv\lib\site-packages\prefect\utilities\importtools.py", line 156, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at 'dbt_flow.py' encountered an exception

Hey @Michael Holvey what version of the prefect-dbt collection do you have installed? I believe this actually should be fixed with 0.2.3 https://github.com/PrefectHQ/prefect-dbt/releases/tag/v0.2.3 as it's likely a result of this issue https://github.com/PrefectHQ/prefect-dbt/issues/67

ahh looks like i have 0.2.2 - let me try upgrading

Hmmmm, are the flows/tasks defined as asynchronous functions? they shouldn't have to be but I'd try testing it that way to see if it works for starters, The error at least previously comes up as a result of calling an asynchronous task within a synchronous flow context. I've been wrestling with a related issue recently that I haven't quite been able to crack just yet but that'd be a good place to start narrowing it down, There's also still this issue in the main prefect repo which could be related as well

wasn't doing anything async before but i just ran it as async with the same errors. not sure if this is because i'm running on windows - wish i had a linux environment to test in!

The methods in the DBT collection are generally asynchronous, most of the time that's not an issue but it has come up before as an issue would you mind posting an minimum example of the for you're running?

of course - my code is really simple, so happy to share! config/profile info is obfuscated.

from prefect import flow
from prefect_dbt.cli import DbtCliProfile
from prefect_dbt.cli.configs import GlobalConfigs, TargetConfigs
from prefect_dbt.cli.commands import trigger_dbt_cli_command
import asyncio

@flow
async def run_dbt():

    target_configs_extras = dict(
        server="SERVER_IP",
        user="USER_ACCT",
        password="USER_PASS",
        port=1433,
        database="ANALYTICS",
    )
    target_configs = TargetConfigs(
        type="sqlserver",
        schema="dbt",
        threads=4,
        extras=target_configs_extras
    )
    dbt_cli_profile = DbtCliProfile(
        name="PROFILE_NAME",
        target="dev",
        target_configs=target_configs,
    )

    profile = dbt_cli_profile

    result = await trigger_dbt_cli_command(command="dbt run",
        project_dir="/dbt_project"
        )
    return result

asyncio.run(run_dbt())

Hmm nope I don't think it's related to what I thought it was, Would you be willing to open a bug on the git hub repo for this issue with your example and the error, also including any potential troubleshooting steps you've take?

yessir will do - in the middle of stuff now but will try to tackle that later tonight. thanks for following up so quickly!

No worries 😄

Hi @Michael Holvey - Can you try running the CLI command to create your deployment from the root directory of your flow’s project?