diff --git a/src/prefect/core/flow.py b/src/prefect/core/flow.py index e259064ba5b8..757dfc7f7861 100644 --- a/src/prefect/core/flow.py +++ b/src/prefect/core/flow.py @@ -1131,7 +1131,7 @@ def serialize(self, build: bool = False) -> dict: if build: if not self.storage: raise ValueError("This flow has no storage to build") - if self not in self.storage: + if self.name not in self.storage: self.storage.add_flow(self) storage = self.storage.build() # type: Optional[Storage] else: diff --git a/tests/core/test_flow.py b/tests/core/test_flow.py index b4b820a27f9f..630d6309d0f2 100644 --- a/tests/core/test_flow.py +++ b/tests/core/test_flow.py @@ -1401,6 +1401,14 @@ def test_serialize_adds_flow_to_storage_if_build(self): s_build = f.serialize(build=True) assert f.name in f.storage + def test_serialize_can_be_called_twice(self): + f = Flow(name="test", storage=prefect.environments.storage.Memory()) + s_no_build = f.serialize() + assert f.name not in f.storage + + f.serialize(build=True) + f.serialize(build=True) + def test_serialize_fails_with_no_storage(self): f = Flow(name="test") with pytest.raises(ValueError):