Put flows in a subdirectory of each integration

celigo/core.py

@@ -1,4 +1,5 @@
 import glob
+import collections
 import logging
 import requests
 import io
@@ -9,6 +10,7 @@ from slugify import slugify
 
 from .prompt import prompt
 
+logging.basicConfig(level=logging.INFO)
 L = logging.getLogger(__name__)
 DEFAULT_BASE_URL = "https://api.integrator.io/v1/"
 
@@ -76,9 +78,10 @@ class BackupCeligo(object):
         self.session = requests.Session()
         self.session.headers.update(self.headers)
 
-    def ensure_directories_exist(self):
+    def ensure_directories_exist(self, subdirs=None):
         """ Make the directory if it doesn't exist """
-        subdirs = ('imports', 'connections')
+        if not subdirs:
+            subdirs = ('integrations', 'connections')
 
         for subdir in subdirs:
             _dir = os.path.join(self.data_dir, subdir)
@@ -123,6 +126,17 @@ class BackupCeligo(object):
             varable.
             Once this is cached, save the imports.
         """
+        try:
+            integrations = self._celigo_api_get("integrations/")
+        except requests.exceptions.RequestException:
+            L.info('HTTP Request failed')
+            raise
+        # Setup integrations dictionaries
+        for integration in integrations:
+            self.imports_cache[integration['_id']] = {
+                    'name': integration['name'],
+                    'slug': slugify(integration['name']),
+                    'flows': []}
         try:
             flows = self._celigo_api_get("flows/")
             for flow in flows:
@@ -131,10 +145,16 @@ class BackupCeligo(object):
             L.info('HTTP Request failed')
             raise
         L.info("Got all imports, writing now")
-        L.info("We have imports for: %s", ", ".join(self.imports_cache.keys()))
 
-        for flow_name, (import_id, import_conf) in self.imports_cache.items():
+        for integration_id, integration in self.imports_cache.items():
-            self.save_import(flow_name, auto)
+            # Make sure the integration directories exist
+            base = "integrations/"+integration_id
+            subdirs = (base + "imports", base + "exports")
+            self.ensure_directories_exist(subdirs=subdirs)
+
+            # save the files
+            for flow in integration['flows']:
+                self.save_flow(integration_id, flow, auto)
 
     def restore(self, auto=False):
         """
@@ -174,28 +194,35 @@ class BackupCeligo(object):
         """
             Stores the import in self.imports_cache before write.
         """
-        flow_name = slugify(flow['name'])
+        flow_name = flow['name']
         import_id = flow['_importId']
+        integration_id = flow['_integrationId']
         import_conf = self._celigo_api_get(
-            "imports/{id}/distributed".format(
+            "imports/{id}/distributed".format(id=import_id))
-                id=import_id))
 
-        self.imports_cache[flow_name] = (import_id, import_conf)
+        self.imports_cache[integration_id]['flows'].append({
+            "name": flow_name,
+            "id": import_id,
+            "configuration": import_conf
+        })
 
-    def save_import(self, flow_name, auto=False):
+    def save_flow(self, integration_id, flow, auto=False):
         """
             Write the import to a .json file  with name_id.json format.
             Prompt for overwrite.
-            :param flow_name: the slugified name of the flow as a key
+            :param flow: dictionary of "name", "id", "configuration" for the
+                flow.
             :param auto: if auto is true, don't prompt for overwrite
         """
-        import_id, import_conf = self.imports_cache[flow_name]
+        flow_name = flow['name']
+        flow_id = flow['id']
+        flow_conf = flow['configuration']
 
-        filename = os.path.join(
+        filename = os.path.join(self.data_dir,
-            self.data_dir,
+                                "integrations",
+                                integration_id,
                                 "imports",
-            "%s_%s.json" % (flow_name, import_id))
+                                "%s_%s.json" % (slugify(flow_name), flow_id))
-
         write = True
 
         # By default, we prompt for overwrites
@@ -209,7 +236,7 @@ class BackupCeligo(object):
             write = bool(overwrite == "Yes")
 
         if write:
-            self.write_json(filename, import_conf)
+            self.write_json(filename, flow_conf)
         else:
             L.info("You chose not to save this file.")
 

tests.py

@@ -48,9 +48,10 @@ class CeligoTest(unittest.TestCase):
         """
             Test that the ensure_directories_exist works properly.
         """
+        fake_id = "fake_integration_id"
         tempdir = tempfile.mkdtemp('celigo_testing')
         bc = celigo.BackupCeligo(tempdir)
-        imports_dir = os.path.join(tempdir, "imports")
+        imports_dir = os.path.join(tempdir, fake_id, "imports")
         connections_dir = os.path.join(tempdir, "connections")
         # Check that the directories don't exist already.
         self.assertFalse(
@@ -59,8 +60,9 @@ class CeligoTest(unittest.TestCase):
         self.assertFalse(
             os.path.exists(connections_dir),
             "connections dir exists")
+
         # Make the directories.
-        bc.ensure_directories_exist()
+        bc.ensure_directories_exist((imports_dir, connections_dir))
         self.assertTrue(
             os.path.exists(imports_dir),
             "Did not create proper directory"
@@ -71,7 +73,7 @@ class CeligoTest(unittest.TestCase):
             )
 
         # Make sure nothing errors if the directories exist already.
-        bc.ensure_directories_exist()
+        bc.ensure_directories_exist((imports_dir, connections_dir))
 
     # @requests_mock.Mocker()
     # def test_fake_requests(self, rqm):