Repository: PrefectHQ/prefect

Issue:
Raise a better error when using custom schedule filters with cloud
Currently custom filters on schedules aren't supported, since the filters run on cloud's servers. Right now the error message raised is a long `ValidationError` traceback that just notes that the function is "invalid" - we should provide a nicer error message to let users know some context around why.

Discussion:

Patch:
diff --git a/changes/pr3540.yaml b/changes/pr3540.yaml
new file mode 100644
index 0000000000..8113eacc34
--- /dev/null
+++ b/changes/pr3540.yaml
@@ -0,0 +1,2 @@
+enhancement:
+  - "Raise a better error message when trying to register a flow with a schedule using custom filter functions - [#3450](https://github.com/PrefectHQ/prefect/pull/3540)"
diff --git a/src/prefect/utilities/serialization.py b/src/prefect/utilities/serialization.py
index 9bf7cca2af..7225c02b8f 100644
--- a/src/prefect/utilities/serialization.py
+++ b/src/prefect/utilities/serialization.py
@@ -382,19 +382,21 @@ class StatefulFunctionReference(fields.Field):
 
         try:
             qual_name = to_qualified_name(value)
-        except Exception as exc:
-            raise ValidationError(
-                f"Invalid function reference, function required, got {value}"
-            ) from exc
-
-        # sort matches such that the longest / most specific match comes first
-        valid_bases = sorted(
-            [fn for fn in self.valid_functions if qual_name.startswith(fn)],
-            key=lambda k: len(k),
-            reverse=True,
-        )
+        except Exception:
+            valid_bases = []
+        else:
+            # sort matches such that the longest / most specific match comes first
+            valid_bases = sorted(
+                [fn for fn in self.valid_functions if qual_name.startswith(fn)],
+                key=lambda k: len(k),
+                reverse=True,
+            )
+
         if not valid_bases and self.reject_invalid:
-            raise ValidationError("Invalid function reference: {}".format(value))
+            raise ValidationError(
+                "When running with Prefect Cloud/Server, custom functions aren't supported "
+                f"for `{type(obj).__name__}.{attr}`. Unable to serialize `{value!r}`."
+            )
         else:
             base_name = next(filter(None, valid_bases)) if valid_bases else qual_name
 


Test patch:
diff --git a/tests/utilities/test_serialization.py b/tests/utilities/test_serialization.py
index 9cbc8b918f..e4313f9404 100644
--- a/tests/utilities/test_serialization.py
+++ b/tests/utilities/test_serialization.py
@@ -240,7 +240,9 @@ class TestStatefulFunctionReferenceField:
         assert serialized["f"]["kwargs"] == {"x": 1, "y": 2, "z": 99}
 
     def test_serialize_invalid_fn(self):
-        with pytest.raises(marshmallow.ValidationError):
+        with pytest.raises(
+            marshmallow.ValidationError, match="custom functions aren't supported"
+        ):
             self.Schema().dump(dict(f=fn2))
 
     def test_serialize_invalid_fn_without_validation(self):
@@ -283,7 +285,9 @@ class TestStatefulFunctionReferenceField:
             def __call__(self, a, b):
                 return a + b
 
-        with pytest.raises(marshmallow.ValidationError, match="function required"):
+        with pytest.raises(
+            marshmallow.ValidationError, match="custom functions aren't supported"
+        ):
             self.Schema().dump(dict(f=Foo()))
 
     def test_serialize_none(self):