Running a deployment with a flow call in it fails

[zanieb]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the Prefect documentation for this issue.
• I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

If a script used by a deployment calls the flow, a flow run will error during loading.

Reproduction

# mre.py
from prefect import flow


@flow
def foo():
    pass


foo()

$ prefect deployment build ./mre.py:foo -n 'mre' --apply
$ prefect deployment run foo/mre
$ prefect agent start -q default

Error

Agent started! Looking for work from queue(s): default...
17:19:47.940 | INFO    | prefect.agent - Submitting flow run 'b51c4799-75a2-4f6d-9b41-c136cd5d2a44'
17:19:48.025 | INFO    | prefect.infrastructure.process - Opening process 'tangerine-ammonite'...
17:19:48.036 | INFO    | prefect.agent - Completed submission of flow run 'b51c4799-75a2-4f6d-9b41-c136cd5d2a44'
17:19:49.953 | INFO    | prefect.engine - Created flow run 'coral-newt' for flow 'foo'
17:19:49.973 | ERROR   | Flow run 'coral-newt' - Crash detected! Execution was interrupted by an unexpected exception.
17:19:49.989 | ERROR   | Flow run 'tangerine-ammonite' - 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 "/private/var/folders/q9/1myqgwxn2r3fvkvf6jy3npcm0000gn/T/tmpmg2og7qfprefect/mre.py", line 10, in <module>
    foo()
  File "/Users/mz/dev/prefect/src/prefect/flows.py", line 384, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/Users/mz/dev/prefect/src/prefect/engine.py", line 158, in enter_flow_run_engine_from_flow_call
    return anyio.run(begin_run)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/Users/mz/dev/prefect/src/prefect/client.py", line 103, in with_injected_client
    return await fn(*args, **kwargs)
  File "/Users/mz/dev/prefect/src/prefect/engine.py", line 231, in create_then_begin_flow_run
    state = await begin_flow_run(
  File "/Users/mz/dev/prefect/src/prefect/engine.py", line 364, in begin_flow_run
    await result_filesystem._save(is_anonymous=True)
  File "/Users/mz/dev/prefect/src/prefect/blocks/core.py", line 717, 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 "/Users/mz/dev/prefect/src/prefect/engine.py", line 256, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/Users/mz/dev/prefect/src/prefect/client.py", line 103, in with_injected_client
    return await fn(*args, **kwargs)
  File "/Users/mz/dev/prefect/src/prefect/deployments.py", line 69, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(import_object, str(import_path))
  File "/Users/mz/dev/prefect/src/prefect/utilities/asyncutils.py", line 57, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/opt/homebrew/Caskroom/miniconda/base/envs/nebula-dev/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/Users/mz/dev/prefect/src/prefect/utilities/importtools.py", line 193, in import_object
    module = load_script_as_module(script_path)
  File "/Users/mz/dev/prefect/src/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 'mre.py' encountered an exception
17:19:50.251 | INFO    | prefect.infrastructure.process - Process 'tangerine-ammonite' exited cleanly.

Versions

Version:             2.2.0+214.g4c6bb7cc2
API version:         0.8.0
Python version:      3.10.4
Git commit:          4c6bb7cc
Built:               Tue, Aug 30, 2022 3:58 PM
OS/Arch:             darwin/x86_64
Profile:             default
Server type:         ephemeral
Server:
  Database:          sqlite
  SQLite version:    3.39.2

Additional context

The flow will also execute when the deployment is built, which is not ideal.

We used to bypass this by short-circuiting flow calls during loading. We can tell when we are loading our own code to inspect the flow object, and can do nothing when the flow is called.

It's likely not worth attempting to fix the exception when this occurs and instead do one of:

• Raise a nicer exception indicating that we found a flow call, this will prevent the deployment from being built
• Ignore flow calls during loading, raise a warning

[anna-geller]

I like those two options:

Raise a nicer exception indicating that we found a flow call, this will prevent the deployment from being built
Ignore flow calls during loading