Merge pull request #1007 from PrefectHQ/storage-update

Add fix for calling serialize twice

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:

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):