Remove cloud result handler

CHANGELOG.md

@@ -36,6 +36,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 - The CLI command `prefect execute-flow` and `prefect execute-cloud-flow` no longer exist - [#1059](https://github.com/PrefectHQ/prefect/pull/1059)
 - The `slack_notifier` state handler now uses a `webhook_secret` kwarg to pull the URL from a Secret - [#1075](https://github.com/PrefectHQ/prefect/issues/1075)
+- Remove the `CloudResultHandler` default result handler - [#1198](https://github.com/PrefectHQ/prefect/pull/1198)
 
 ### Contributors
 

docs/outline.toml

@@ -126,7 +126,7 @@ classes = ["JSONResultHandler", "GCSResultHandler", "LocalResultHandler", "S3Res
 [pages.engine.cloud]
 title = "Cloud"
 module = "prefect.engine.cloud"
-classes = ["CloudFlowRunner", "CloudTaskRunner", "CloudResultHandler"]
+classes = ["CloudFlowRunner", "CloudTaskRunner"]
 
 [pages.environments.storage]
 title = "Storage"

src/prefect/config.toml

@@ -70,7 +70,7 @@ run_on_schedule = true
 
     [engine.result_handler]
     # the default task runner, specified using a full path
-    default_class = "prefect.engine.cloud.CloudResultHandler"
+    default_class = ""
 
     [engine.task_runner]
     # the default task runner, specified using a full path

src/prefect/engine/__init__.py

@@ -86,18 +86,17 @@ def get_default_result_handler_class() -> type:
     value is a string, it will attempt to load the already-imported object. Otherwise, the
     value is returned.
 
-    Defaults to `CloudResultHandler` if the string config value can not be loaded
+    Defaults to `None` if the string config value can not be loaded
     """
     config_value = config.get_nested("engine.result_handler.default_class")
 
     if isinstance(config_value, str):
+        if not config_value:
+            return lambda *args, **kwargs: None  # type: ignore
         try:
             return prefect.utilities.serialization.from_qualified_name(config_value)
         except ValueError:
-            warn(
-                "Could not import {}; using "
-                "prefect.engine.cloud.CloudResultHandler instead.".format(config_value)
-            )
-            return prefect.engine.cloud.CloudResultHandler
+            warn("Could not import {}; using " "None instead.".format(config_value))
+            return lambda *args, **kwargs: None  # type: ignore
     else:
         return config_value

src/prefect/engine/cloud/__init__.py

@@ -1,3 +1,2 @@
-from prefect.engine.cloud.result_handler import CloudResultHandler
 from prefect.engine.cloud.task_runner import CloudTaskRunner
 from prefect.engine.cloud.flow_runner import CloudFlowRunner

src/prefect/serialization/result_handlers.py

@@ -3,7 +3,6 @@
 
 from marshmallow import ValidationError, fields, post_load
 
-from prefect.engine.cloud.result_handler import CloudResultHandler
 from prefect.engine.result_handlers import (
     GCSResultHandler,
     JSONResultHandler,
@@ -39,13 +38,6 @@ def create_object(self, data: dict, **kwargs: Any) -> None:
         return None
 
 
-class CloudResultHandlerSchema(BaseResultHandlerSchema):
-    class Meta:
-        object_class = CloudResultHandler
-
-    result_handler_service = fields.String(allow_none=True)
-
-
 class GCSResultHandlerSchema(BaseResultHandlerSchema):
     class Meta:
         object_class = GCSResultHandler
@@ -84,7 +76,6 @@ class ResultHandlerSchema(OneOfSchema):
         "ResultHandler": BaseResultHandlerSchema,
         "GCSResultHandler": GCSResultHandlerSchema,
         "S3ResultHandler": S3ResultHandlerSchema,
-        "CloudResultHandler": CloudResultHandlerSchema,
         "JSONResultHandler": JSONResultHandlerSchema,
         "LocalResultHandler": LocalResultHandlerSchema,
     }

tests/engine/cloud/test_cloud_task_runner.py

@@ -12,7 +12,7 @@
 from prefect.client import Client
 from prefect.core import Edge, Task
 from prefect.engine.cache_validators import all_inputs
-from prefect.engine.cloud import CloudResultHandler, CloudTaskRunner
+from prefect.engine.cloud import CloudTaskRunner
 from prefect.engine.result import NoResult, Result, SafeResult
 from prefect.engine.result_handlers import (
     JSONResultHandler,

tests/engine/test_defaults.py

@@ -89,7 +89,7 @@ def test_default_task_runner_with_bad_config():
 
 
 def test_default_result_handler():
-    assert engine.get_default_result_handler_class() is engine.cloud.CloudResultHandler
+    assert engine.get_default_result_handler_class()() is None
 
 
 def test_default_result_handler_responds_to_config():
@@ -121,7 +121,4 @@ def test_default_result_handler_with_bad_config():
         {"engine.result_handler.default_class": "prefect.engine. bad import path"}
     ):
         with pytest.warns(UserWarning):
-            assert (
-                engine.get_default_result_handler_class()
-                is engine.cloud.CloudResultHandler
-            )
+            assert engine.get_default_result_handler_class()() is None

tests/serialization/test_result_handlers.py

@@ -4,7 +4,6 @@
 
 import prefect
 from prefect.client import Client
-from prefect.engine.cloud.result_handler import CloudResultHandler
 from prefect.engine.result_handlers import (
     GCSResultHandler,
     JSONResultHandler,
@@ -122,50 +121,6 @@ def test_deserialize_local_result_handler(self, dir):
         assert obj.dir == dir
 
 
-class TestCloudResultHandler:
-    def test_serialize_with_no_attributes(self):
-        with set_temporary_config({"cloud.result_handler": "website"}):
-            serialized = ResultHandlerSchema().dump(CloudResultHandler())
-        assert isinstance(serialized, dict)
-        assert serialized["type"] == "CloudResultHandler"
-        assert serialized["result_handler_service"] == "website"
-        assert "client" not in serialized
-
-    def test_serialize_with_attributes(self):
-        handler = CloudResultHandler(result_handler_service="http://foo.bar")
-        handler.client = Client()
-        serialized = ResultHandlerSchema().dump(handler)
-        assert isinstance(serialized, dict)
-        assert serialized["type"] == "CloudResultHandler"
-        assert serialized["result_handler_service"] == "http://foo.bar"
-        assert "client" not in serialized
-
-    def test_deserialize_cloud_result_handler(self):
-        schema = ResultHandlerSchema()
-        handler = CloudResultHandler(result_handler_service="http://foo.bar")
-        handler._client = Client()
-        obj = schema.load(schema.dump(handler))
-        assert isinstance(obj, CloudResultHandler)
-        assert hasattr(obj, "logger")
-        assert obj.logger.name == "prefect.CloudResultHandler"
-        assert obj.result_handler_service == "http://foo.bar"
-        assert obj._client is None
-
-    def test_deserialize_cloud_result_handler_with_None_populates_from_config(self):
-        schema = ResultHandlerSchema()
-        handler = CloudResultHandler()
-        handler.result_handler_service = None
-        handler._client = Client()
-        serialized = schema.dump(handler)
-        with set_temporary_config({"cloud.result_handler": "new-service"}):
-            obj = schema.load(serialized)
-        assert isinstance(obj, CloudResultHandler)
-        assert hasattr(obj, "logger")
-        assert obj.logger.name == "prefect.CloudResultHandler"
-        assert obj.result_handler_service == "new-service"
-        assert obj._client is None
-
-
 @pytest.mark.xfail(raises=ImportError, reason="google extras not installed.")
 class TestGCSResultHandler:
     def test_serialize(self):