Dependency on import order in user-code

[cicdw]

Bug summary

The following code errors out:

from prefect.states import State
State(type="CANCELLED").to_state_create()

with traceback:

---------------------------------------------------------------------------
PydanticUserError                         Traceback (most recent call last)
Cell In[1], line 2
      1 from prefect.states import State
----> 2 State(type="CANCELLED").to_state_create()

File ~/dev/prefect/src/prefect/client/schemas/objects.py:335, in State.to_state_create(self)
    332 else:
    333     data = None
--> 335 return StateCreate(
    336     type=self.type,
    337     name=self.name,
    338     message=self.message,
    339     data=data,
    340     state_details=self.state_details,
    341 )

    [... skipping hidden 1 frame]

File ~/miniconda3/envs/prefect/lib/python3.10/site-packages/pydantic/_internal/_mock_val_ser.py:47, in MockValSer.__getattr__(self, item)
     45 # raise an AttributeError if `item` doesn't exist
     46 getattr(self._val_or_ser, item)
---> 47 raise PydanticUserError(self._error_message, code=self._code)

PydanticUserError: `StateCreate` is not fully defined; you should define all referenced types, then call `StateCreate.model_rebuild()`.

For further information visit https://errors.pydantic.dev/2.7/u/class-not-fully-defined

It can be resolved by importing anything from the main prefect module:

from prefect import State # any other object would work too

State(type="CANCELLED").to_state_create() # ✅

Version info

Version:             3.1.1.dev6+1.g2624409ffc
API version:         0.8.4
Python version:      3.10.13
Git commit:          2624409f
Built:               Fri, Nov 8, 2024 8:12 AM
OS/Arch:             darwin/arm64
Profile:             ephemeral
Server type:         ephemeral
Pydantic version:    2.7.1
Server:
  Database:          sqlite
  SQLite version:    3.41.2
Integrations:
  prefect-aws:       0.5.0

Additional context

This is related to our known circular import problems; I'm opening this for other users who may be searching with this error message.

[bkempe]

I've seen this in Prefect (3.1.4) logs after an out of memory exception

12:25:42 PM INFO Opening process...
12:25:44 PM INFO Downloading flow code from storage at '.'
12:25:47 PM INFO Process for flow run '[...]' exited with status code: -9; This indicates that the process exited due to a SIGKILL signal. Typically, this is either caused by manual cancellation or high memory usage causing the operating system to terminate the process.
12:26:32 PM INFO Failed to update state of flow run '191eab32-30b3-49fe-8f53-d9561d8ec1a3'
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/prefect/runner/runner.py", line 1159, in _propose_crashed_state
    state = await propose_state(
            ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/engine.py", line 429, in propose_state
    response = await set_state_and_handle_waits(set_state)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/engine.py", line 413, in set_state_and_handle_waits
    response = await set_state_func()
               ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/client/orchestration.py", line 2173, in set_flow_run_state
    state_create = state.to_state_create()
                   ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/client/schemas/objects.py", line 336, in to_state_create
    return StateCreate(
           ^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pydantic/main.py", line 212, in __init__
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pydantic/_internal/_mock_val_ser.py", line 99, in __getattr__
    raise PydanticUserError(self._error_message, code=self._code)
pydantic.errors.PydanticUserError: `StateCreate` is not fully defined; you should define all referenced types, then call `StateCreate.model_rebuild()`.

@zzstoatzz Do you know a workaround here, since the code in question is in Prefect, not our codebase.

Especially concerning here is that "Failed to update state of flow run" indicates that the flow run stays in "Running" state, i.e. occupies concurrency slots.

[EmilRex]

Also seeing this, but when using the API client:

Reproduction

from prefect.client.orchestration import get_client

with get_client(sync_client=True) as client:
    flow_runs = client.read_flow_runs()

Error

$ python repro.py
Traceback (most recent call last):
  File "/Users/emilchristensen/github/examples/repro.py", line 5, in <module>
    flow_runs = client.read_flow_runs()
                ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/emilchristensen/github/examples/.venv/lib/python3.12/site-packages/prefect/client/orchestration.py", line 4065, in read_flow_runs
    return pydantic.TypeAdapter(list[FlowRun]).validate_python(response.json())
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/emilchristensen/github/examples/.venv/lib/python3.12/site-packages/pydantic/type_adapter.py", line 412, in validate_python
    return self.validator.validate_python(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/emilchristensen/github/examples/.venv/lib/python3.12/site-packages/pydantic/_internal/_mock_val_ser.py", line 100, in __getattr__
    raise PydanticUserError(self._error_message, code=self._code)
pydantic.errors.PydanticUserError: `TypeAdapter[list[prefect.client.schemas.objects.FlowRun]]` is not fully defined; you should define `list[prefect.client.schemas.objects.FlowRun]` and all referenced types, then call `.rebuild()` on the instance.

For further information visit https://errors.pydantic.dev/2.10/u/class-not-fully-defined

Version

Version:             3.1.8
API version:         0.8.4
Python version:      3.12.7
Git commit:          53a83ebc
Built:               Tue, Dec 17, 2024 10:20 AM
OS/Arch:             darwin/arm64
Profile:             sandbox
Server type:         cloud
Pydantic version:    2.10.4
Integrations:
  prefect-docker:    0.6.2

$ uv pip freeze | grep pydantic
pydantic==2.10.4
pydantic-core==2.27.2
pydantic-extra-types==2.10.1
pydantic-settings==2.7.0

[zzstoatzz]

hi @bkempe and @EmilRex the workaround is to import prefect.main before using the client

this is a symptom of a larger issue that I've started #16387 to discuss

[KK7NZY]

@zzstoatzz

Hi! I recently encountered this issue locally with a flow that has been running smoothly for about three months. After an update, I'm now seeing the following error:

Failed to process backup: PydanticUserError('EngineContext is not fully defined; you should define Flow, then call EngineContext.model_rebuild().')

Not sure if it is related to this one.

[zzstoatzz]

hi @KK7NZY - its probably tangentially related at least

can you give some more context on when you hit this? Coincidentally, I just fixee at least one case of this here

[KK7NZY]

@zzstoatzz The issue just started appearing after my return to work this week. As far as I know, nothing has changed with the dependencies or related code. We did update the Helm charts to use the latest release of Prefect Server and bumped the Prefect versions, but none of the code itself has been modified.

Apologies for not having more information—it only starts erroring when the flow is running.

The flow uses the DaskTaskRunner, but as I mentioned, this has been working without issues for a while.

The line where it fails is:

@task
def process_backup(...) ....

@flow(task_runner=DaskTaskRunner(cluster_kwargs={"n_workers": 3})
def main_flow()
   ....
    download_futures = [process_backup.submit(...), for ...]

   ....

If I skip this line or assign an empty array, the main flow runs fine. The error only occurs when the subtasks begin to run, with each task raising an error.

....
Failed to process ... chunk: PydanticUserError('`EngineContext` is not fully defined; you should define `Flow`, then call `EngineContext.model_rebuild()`.')
...

Encountered exception during execution: Exception('4 tasks failed during the flow.')
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/flow_engine.py", line 731, in run_context
    yield self
  File "/usr/local/lib/python3.11/site-packages/prefect/flow_engine.py", line 1338, in run_flow_sync
    engine.call_flow_fn()
  File "/usr/local/lib/python3.11/site-packages/prefect/flow_engine.py", line 754, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/callables.py", line 208, in call_with_parameters
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/code/prefect/flows/my_flow.py", line 381, in sync_vuln_metadata
    raise Exception(f"{failures} tasks failed during the flow.")
Exception: 4 tasks failed during the flow.