Merge pull request #36 from ecmwf/feature/deployment

Merge changes from PR #16

docs/content/introductory-course/getting-started.rst

@@ -52,7 +52,7 @@ Copy the following code into a new file called ``hello_pyflow.py``:
    s.check_definition()
    print(s)
 
-   s.deploy_suite(overwrite=True)
+   s.deploy_suite()
    s.replace_on_server(server_host, server_port)
 
 

pyflow/deployment.py

@@ -12,17 +12,15 @@ class DeploymentError(RuntimeError):
 
 
 class Deployment:
-    def __init__(self, suite, headers=True, overwrite=False):
+    def __init__(self, suite, headers=True):
         """
         Base class for all deployments.
 
         Parameters:
             suite(Suite_): The suite object to deploy.
             headers(bool): Whether to deploy the headers.
-            overwrite(bool): Whether to overwrite existing target.
         """
 
-        self._overwrite = overwrite
         self._headers = headers
         self._includes = set()
 
@@ -138,9 +136,6 @@ def deploy_headers(self):
         for inc in self._includes:
             inc.install(where)
 
-    def create_directories(self, path):
-        raise NotImplementedError
-
 
 class Notebook(Deployment, FileListHTMLWrapper):
     """
@@ -186,9 +181,6 @@ def save(self, source, target):
         super().save(source, target)
         self._content.append((target, source))
 
-    def create_directories(self, path):
-        pass
-
 
 class Dummy:
     def copy(*args):
@@ -197,9 +189,6 @@ def copy(*args):
     def save(*args):
         pass
 
-    def create_directories(*args):
-        pass
-
 
 class FileSystem(Deployment):
     """
@@ -215,6 +204,7 @@ class FileSystem(Deployment):
     def __init__(self, suite, path=None, **kwargs):
         super().__init__(suite, **kwargs)
         self.path = path
+        self._processed = set()
 
     def patch_path(self, path):
         """
@@ -232,59 +222,65 @@ def create_directory(self, path):
             except Exception:
                 print("WARNING: Couldn't create directory: {}".format(path))
 
-    def check(self, target, content):
+    def check(self, target):
+        """
+        Check if the target should be deployed.
+        Returns False if target has already been deployed, True otherwise.
+
+        Parameters
+        ----------
+        target : str
+            The target path for deployment.
+
+        Returns
+        -------
+        bool
+            True if the target path is valid for deployment, False otherwise.
+        """
         if target is None:
             raise RuntimeError(
                 "None is not a valid path for deployment. Most likely files/ECF_FILES unspecified"
             )
 
-        if not os.path.exists(target):
+        # if script with same content already deployed, skip
+        if os.path.exists(target):
+            if self.duplicate_write_check(target):
+                return False
+        else:
             self.create_directory(os.path.dirname(target))
-            return True
 
-        previous = open(target, "r").read()
-
-        if previous == content:
-            return False
-
-        if not self._overwrite:
-            raise RuntimeError("File %s exists, not overwriting." % (target,))
-
-        print("Overwriting existing file: %s" % (target,))
         return True
 
     def copy(self, source, target):
         target = self.patch_path(target)
         super().copy(source, target)
 
-        content = open(source, "r").read()
-
-        if not self.check(target, content):
+        if not self.check(target):
             return
 
         print("Copy %s to %s" % (source, target))
 
-        with open(target, "w") as g:
-            g.write(content)
+        with open(source, "r") as f:
+            with open(target, "w") as g:
+                g.write(f.read())
 
     def save(self, source, target):
         target = self.patch_path(target)
         super().save(source, target)
 
-        content = "\n".join(source) if isinstance(source, list) else source
-
-        if not self.check(target, content):
+        if not self.check(target):
             return
 
-        assert isinstance(content, (str, bytes))
-        with open(target, "w" if isinstance(content, str) else "wb") as g:
-            g.write(content)
+        output = "\n".join(source) if isinstance(source, list) else source
+        assert isinstance(output, (str, bytes))
+        with open(target, "w" if isinstance(output, str) else "wb") as g:
+            g.write(output)
 
-    def create_directories(self, path):
-        pass
-        # self.create_directory(self._home + path)
-        # It isn't the job of pyflow to create these.
-        # self.create_directory(self._out + path)
+    def duplicate_write_check(self, target):
+        if target in self._processed:
+            return True
+        self._processed.add(target)
+        return False
 
 
 class DeployGitRepo(FileSystem):

pyflow/host.py

@@ -7,6 +7,7 @@
 import pwd
 import shutil
 import textwrap
+import warnings
 
 from .attributes import Label, Limit
 from .base import STACK
@@ -246,7 +247,12 @@ def build_label(self):
             return Label("exec_host", self.hostname)
 
     def script_submit_arguments(self, submit_arguments):
-        assert len(submit_arguments) == 0
+        if len(submit_arguments) > 0:
+            warnings.formatwarning = lambda mess, *args, **kwargs: "%s" % mess
+            warnings.warn(
+                f"Host {self.__class__.__name__} does not support scheduler submission arguments. "
+                "They will be ignore for script generation"
+            )
         return []
 
     def preamble_init(self, ecflowpath):

tutorials/course/course.ipynb

@@ -156,7 +156,7 @@
     }
    ],
    "source": [
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -388,7 +388,7 @@
     "        MyTask('Five', 5)\n",
     "    \n",
     "print(s)\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -484,7 +484,7 @@
     "with pf.Suite('CountingSuite', files=os.path.join(filesdir, 'CountingSuite')) as s:\n",
     "    print(MyFamily('TaskCounter', 7))\n",
     "\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -606,7 +606,7 @@
     "    MyFamily('fam2', 5)\n",
     "    \n",
     "print(s)\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -1087,7 +1087,7 @@
     "print(s.prod)\n",
     "print('\\n------------------------------------\\n')\n",
     "print(s.dev)\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -1568,7 +1568,7 @@
     "            LsTest()\n",
     "        )\n",
     "    \n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -1748,7 +1748,7 @@
     "            pf.sequence(DelegatingTest(test_cfg) for test_cfg in config.tests)\n",
     "            \n",
     "s = DelegatedSuite(CombinedConfig())\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -2271,7 +2271,7 @@
     "        \n",
     "print(s)\n",
     "print(\"script:\\n\", f.uses_var.script, '\\n')\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -2329,7 +2329,7 @@
     "        \n",
     "print(s)\n",
     "print(\"script:\\n\", f.uses_var.script, '\\n')\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -2410,7 +2410,7 @@
     "print(s)\n",
     "print(\"script external:\\n\", f.external_variable.script, '\\n')\n",
     "print(\"script default:\\n\", f.default_value.script, '\\n')\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -2478,7 +2478,7 @@
     "                WaitSeconds(2)\n",
     "            )\n",
     "\n",
-    "s.deploy_suite(overwrite=True)\n",
+    "s.deploy_suite()\n",
     "s.replace_on_server(server_host, server_port)"
    ]
   },
@@ -3896,9 +3896,9 @@
     "s2 = NestedLoopingSuite(Config2())\n",
     "print(s2)\n",
     "\n",
-    "s1.deploy_suite(overwrite=True)\n",
+    "s1.deploy_suite()\n",
     "s1.replace_on_server(server_host, server_port)\n",
-    "s2.deploy_suite(overwrite=True)\n",
+    "s2.deploy_suite()\n",
     "s2.replace_on_server(server_host, server_port)"
    ]
   }