PrefectHQ / prefect

Prefect is a workflow orchestration framework for building resilient data pipelines in Python.
https://prefect.io
Apache License 2.0
16.11k stars 1.57k forks source link

Error while creating StateCreate object in prefect python client #15646

Open gigaverse-oz opened 6 days ago

gigaverse-oz commented 6 days ago

Bug summary

Getting Pydantic error when trying to create StateCreate object (used by state.to_state_create()): pydantic.errors.PydanticUserError:StateCreateis not fully defined; you should define all referenced types, then callStateCreate.model_rebuild().

User level code that fails:

async with get_client() as client:
        # response = await client.hello()

        object_to_parse = {"timestamp": datetime.now(timezone.utc).isoformat()}
        run_params = {"event_details": json.dumps(object_to_parse)}
        run_schedule = Scheduled(scheduled_time=datetime.now(timezone.utc)+timedelta(minutes=1))
        response = await client.create_flow_run_from_deployment(deployment_id="907b9cc4-07b7-4150-8201-5f5d67b421bc",
                                                                state=run_schedule,parameters=run_params) # RAISE PYDANTIC EXCEPTION
        print(response.state.is_scheduled())

Version info (prefect version output)

Version:             3.0.4
API version:         0.8.4
Python version:      3.11.6
Git commit:          c068d7e2
Built:               Tue, Oct 1, 2024 11:54 AM
OS/Arch:             linux/x86_64
Profile:             local
Server type:         server
Pydantic version:    2.9.2
Integrations:
  prefect-kubernetes: 0.5.0

(Was tested with prefect versions 3.0.3,3.0.6 and 3.0.7 as well)

Additional context

In the actions.py the following exist:

if TYPE_CHECKING: # = False
    from prefect.results import BaseResult, ResultRecordMetadata

R = TypeVar("R")

class StateCreate(ActionBaseModel):
    """Data used by the Prefect REST API to create a new state."""

    type: StateType
    name: Optional[str] = Field(default=None)
    message: Optional[str] = Field(default=None, examples=["Run started"])
    state_details: StateDetails = Field(default_factory=StateDetails)
    data: Union["BaseResult[R]", "ResultRecordMetadata", Any] = Field( # The classes were not imported so pydantic failed
        default=None,
    )

when changed TYPE_CHECKING=TRUE, the entire import of the prefect library fails:

ImportError: cannot import name 'DEFAULT_BLOCK_SCHEMA_VERSION' from partially initialized module 'prefect.client.schemas' (most likely due to a circular import) (/workspaces/prefect-poc/.venv/lib/python3.11/site-packages/prefect/client/schemas/__init__.py)

Every thing worked for me when changed the class definition to:

class StateCreate(ActionBaseModel):
    """Data used by the Prefect REST API to create a new state."""

    type: StateType
    name: Optional[str] = Field(default=None)
    message: Optional[str] = Field(default=None, examples=["Run started"])
    state_details: StateDetails = Field(default_factory=StateDetails)
    data: Union[Any] = Field( 
        default=None,
    )
zzstoatzz commented 6 days ago

hi @gigaverse-oz - can you try upgrading? this appears to be the same thing we addressed in https://github.com/PrefectHQ/prefect/pull/15615