Remove deprecated result fields and classes and associated utilities (#…

…16552)

src/prefect/client/schemas/actions.py

@@ -39,7 +39,7 @@
 from prefect.utilities.pydantic import get_class_fields_only
 
 if TYPE_CHECKING:
-    from prefect.results import BaseResult, ResultRecordMetadata
+    from prefect.results import ResultRecordMetadata
 
 R = TypeVar("R")
 
@@ -51,7 +51,7 @@ class StateCreate(ActionBaseModel):
     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[Any]", "ResultRecordMetadata", Any] = Field(
+    data: Union["ResultRecordMetadata", Any] = Field(
         default=None,
     )
 

src/prefect/client/schemas/objects.py

@@ -10,7 +10,6 @@
     Generic,
     Optional,
     Union,
-    cast,
     overload,
 )
 from uuid import UUID, uuid4
@@ -65,7 +64,7 @@
 
 if TYPE_CHECKING:
     from prefect.client.schemas.actions import StateCreate
-    from prefect.results import BaseResult, ResultRecordMetadata
+    from prefect.results import ResultRecordMetadata
 
     DateTime = pendulum.DateTime
 else:
@@ -195,9 +194,7 @@ class StateDetails(PrefectBaseModel):
 
 
 def data_discriminator(x: Any) -> str:
-    if isinstance(x, dict) and "type" in x and x["type"] != "unpersisted":
-        return "BaseResult"
-    elif isinstance(x, dict) and "storage_key" in x:
+    if isinstance(x, dict) and "storage_key" in x:
         return "ResultRecordMetadata"
     return "Any"
 
@@ -214,7 +211,6 @@ class State(ObjectBaseModel, Generic[R]):
     state_details: StateDetails = Field(default_factory=StateDetails)
     data: Annotated[
         Union[
-            Annotated["BaseResult[R]", Tag("BaseResult")],
             Annotated["ResultRecordMetadata", Tag("ResultRecordMetadata")],
             Annotated[Any, Tag("Any")],
         ],
@@ -347,14 +343,11 @@ def to_state_create(self) -> "StateCreate":
         """
         from prefect.client.schemas.actions import StateCreate
         from prefect.results import (
-            BaseResult,
             ResultRecord,
             should_persist_result,
         )
 
-        if isinstance(self.data, BaseResult):
-            data = cast(BaseResult[R], self.data)
-        elif isinstance(self.data, ResultRecord) and should_persist_result():
+        if isinstance(self.data, ResultRecord) and should_persist_result():
             data = self.data.metadata
         else:
             data = None

src/prefect/deployments/flow_runs.py

@@ -10,7 +10,7 @@
 from prefect.client.utilities import inject_client
 from prefect.context import FlowRunContext, TaskRunContext
 from prefect.logging import get_logger
-from prefect.results import BaseResult, ResultRecordMetadata
+from prefect.results import ResultRecordMetadata
 from prefect.states import Pending, Scheduled
 from prefect.tasks import Task
 from prefect.telemetry.run_telemetry import (
@@ -25,7 +25,6 @@
 
 prefect.client.schemas.StateCreate.model_rebuild(
     _types_namespace={
-        "BaseResult": BaseResult,
         "ResultRecordMetadata": ResultRecordMetadata,
     }
 )

src/prefect/flow_engine.py

@@ -55,7 +55,6 @@
     patch_print,
 )
 from prefect.results import (
-    BaseResult,
     ResultStore,
     get_result_store,
     should_persist_result,
@@ -307,10 +306,7 @@ def result(self, raise_on_failure: bool = True) -> "Union[R, State, None]":
         if self._return_value is not NotSet and not isinstance(
             self._return_value, State
         ):
-            if isinstance(self._return_value, BaseResult):
-                _result = self._return_value.get()
-            else:
-                _result = self._return_value
+            _result = self._return_value
 
             if asyncio.iscoroutine(_result):
                 # getting the value for a BaseResult may return an awaitable
@@ -873,10 +869,7 @@ async def result(self, raise_on_failure: bool = True) -> "Union[R, State, None]"
         if self._return_value is not NotSet and not isinstance(
             self._return_value, State
         ):
-            if isinstance(self._return_value, BaseResult):
-                _result = self._return_value.get()
-            else:
-                _result = self._return_value
+            _result = self._return_value
 
             if asyncio.iscoroutine(_result):
                 # getting the value for a BaseResult may return an awaitable

src/prefect/main.py

@@ -8,7 +8,7 @@
 from prefect.tasks import task, Task
 from prefect.context import tags
 from prefect.utilities.annotations import unmapped, allow_failure
-from prefect.results import BaseResult, ResultRecordMetadata
+from prefect.results import ResultRecordMetadata
 from prefect.flow_runs import pause_flow_run, resume_flow_run, suspend_flow_run
 from prefect.client.orchestration import get_client
 from prefect.client.cloud import get_cloud_client
@@ -29,7 +29,6 @@
 _types: dict[str, Any] = dict(
     Task=Task,
     Flow=Flow,
-    BaseResult=BaseResult,
     ResultRecordMetadata=ResultRecordMetadata,
 )
 prefect.context.FlowRunContext.model_rebuild(_types_namespace=_types)