Add support for dedicated storage handler

datadog_sync/commands/shared/utils.py

@@ -1,10 +1,8 @@
 import asyncio
-import os
 from sys import exit
 
-from datadog_sync.constants import DESTINATION_ORIGIN, DESTINATION_RESOURCES_DIR, SOURCE_RESOURCES_DIR, Command
+from datadog_sync.constants import Command
 from datadog_sync.utils.configuration import Configuration, build_config
-from datadog_sync.utils.resource_utils import dump_resources
 from datadog_sync.utils.resources_handler import ResourcesHandler
 
 
@@ -21,8 +19,7 @@ def run_cmd(cmd: Command, **kwargs):
         cfg.logger.error("Process interrupted by user")
         if cmd == Command.SYNC:
             cfg.logger.info("Writing synced resources to disk before exit...")
-            synced_resource_types = set(handler.resources_manager.all_resources_to_type.values())
-            dump_resources(cfg, synced_resource_types, DESTINATION_ORIGIN)
+            cfg.state.dump_state()
             exit(0)
 
     if cfg.logger.exception_logged:
@@ -39,10 +36,8 @@ async def run_cmd_async(cfg: Configuration, handler: ResourcesHandler, cmd: Comm
 
         # Run specific handler
         if cmd == Command.IMPORT:
-            os.makedirs(SOURCE_RESOURCES_DIR, exist_ok=True)
             await handler.import_resources()
         elif cmd == Command.SYNC:
-            os.makedirs(DESTINATION_RESOURCES_DIR, exist_ok=True)
             await handler.apply_resources()
         elif cmd == Command.DIFFS:
             await handler.diffs()

datadog_sync/constants.py

@@ -23,14 +23,8 @@
 
 # Default variables
 DEFAULT_API_URL = "https://api.datadoghq.com"
-RESOURCES_DIR = "resources/"
-RESOURCE_FILE_PATH = "resources/{}/{}.json"
-SOURCE_RESOURCES_DIR = "resources/source"
-DESTINATION_RESOURCES_DIR = "resources/destination"
 
 LOGGER_NAME = "datadog_sync_cli"
-SOURCE_ORIGIN = "source"
-DESTINATION_ORIGIN = "destination"
 VALIDATE_ENDPOINT = "/api/v1/validate"
 
 # Bool constants
@@ -44,3 +38,10 @@ class Command(Enum):
     IMPORT = "import"
     SYNC = "sync"
     DIFFS = "diffs"
+
+
+# Origin
+class Origin(Enum):
+    ALL = "all"
+    SOURCE = "source"
+    DESTINATION = "destination"

datadog_sync/model/authn_mappings.py

@@ -56,7 +56,7 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
 
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
-        d_id = self.resource_config.destination_resources[_id]["id"]
+        d_id = self.config.state.destination[self.resource_type][_id]["id"]
         resource["id"] = d_id
         payload = {"data": resource}
         resp = await destination_client.patch(self.resource_config.base_path + f"/{d_id}", payload)
@@ -67,7 +67,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/dashboard_lists.py

@@ -71,32 +71,32 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         dashboards = copy.deepcopy(resource["dashboards"])
         dash_list_diff = check_diff(
             self.resource_config,
-            self.resource_config.destination_resources[_id]["dashboards"],
+            self.config.state.destination[self.resource_type][_id]["dashboards"],
             dashboards,
         )
         resource.pop("dashboards")
 
         resp = await destination_client.put(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             resource,
         )
 
         resp.pop("dashboards")
-        self.resource_config.destination_resources[_id].update(resp)
+        self.config.state.destination[self.resource_type][_id].update(resp)
 
         if dash_list_diff:
             await self.update_dash_list_items(
-                self.resource_config.destination_resources[_id]["id"],
+                self.config.state.destination[self.resource_type][_id]["id"],
                 dashboards,
-                self.resource_config.destination_resources[_id],
+                self.config.state.destination[self.resource_type][_id],
             )
 
-        return _id, self.resource_config.destination_resources[_id]
+        return _id, self.config.state.destination[self.resource_type][_id]
 
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/dashboards.py

@@ -55,7 +55,7 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         resp = await destination_client.put(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             resource,
         )
 
@@ -64,7 +64,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/downtime_schedules.py

@@ -52,7 +52,7 @@ async def import_resource(self, _id: Optional[str] = None, resource: Optional[Di
         return str(resource["id"]), resource
 
     async def pre_resource_action_hook(self, _id, resource: Dict) -> None:
-        if _id not in self.resource_config.destination_resources:
+        if _id not in self.config.state.destination[self.resource_type]:
             schedule = resource["attributes"].get("schedule")
             if schedule and "start" in schedule:
                 current_time = datetime.utcnow()
@@ -69,7 +69,7 @@ async def pre_resource_action_hook(self, _id, resource: Dict) -> None:
             # this is to avoid unnecessary diff outputs
             if resource["attributes"].get("schedule"):
                 one_time_source = resource["attributes"].get("schedule")
-                one_time_created = self.resource_config.destination_resources[_id]["attributes"].get("schedule")
+                one_time_created = self.config.state.destination[self.resource_type][_id]["attributes"].get("schedule")
                 if one_time_created.get("start") and one_time_source.get("start"):
                     start_source = parse(one_time_source["start"])
                     start_created = parse(one_time_created["start"])
@@ -93,10 +93,10 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
 
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
-        resource["id"] = self.resource_config.destination_resources[_id]["id"]
+        resource["id"] = self.config.state.destination[self.resource_type][_id]["id"]
         payload = {"data": resource}
         resp = await destination_client.patch(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             payload,
         )
 
@@ -105,7 +105,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/downtimes.py

@@ -62,7 +62,7 @@ async def import_resource(self, _id: Optional[str] = None, resource: Optional[Di
         return str(resource["id"]), resource
 
     async def pre_resource_action_hook(self, _id, resource: Dict) -> None:
-        if _id not in self.resource_config.destination_resources:
+        if _id not in self.config.state.destination[self.resource_type]:
             current_time = round(datetime.now().timestamp())
             if resource["recurrence"] is None:
                 # If the downtime start time is in the past, convert it to now + 1min
@@ -81,12 +81,12 @@ async def pre_resource_action_hook(self, _id, resource: Dict) -> None:
         else:
             # If start or end times of the resource are in the past, we set to the current destination `start` and `end`
             # this is to avoid unnecessary diff outputs
-            if resource.get("start") and self.resource_config.destination_resources[_id].get("start"):
-                if resource["start"] < self.resource_config.destination_resources[_id]["start"]:
-                    resource["start"] = self.resource_config.destination_resources[_id]["start"]
-            if resource.get("end") and self.resource_config.destination_resources[_id].get("end"):
-                if resource["end"] < self.resource_config.destination_resources[_id]["end"]:
-                    resource["end"] = self.resource_config.destination_resources[_id]["end"]
+            if resource.get("start") and self.config.state.destination[self.resource_type][_id].get("start"):
+                if resource["start"] < self.config.state.destination[self.resource_type][_id]["start"]:
+                    resource["start"] = self.config.state.destination[self.resource_type][_id]["start"]
+            if resource.get("end") and self.config.state.destination[self.resource_type][_id].get("end"):
+                if resource["end"] < self.config.state.destination[self.resource_type][_id]["end"]:
+                    resource["end"] = self.config.state.destination[self.resource_type][_id]["end"]
 
     async def pre_apply_hook(self) -> None:
         pass
@@ -100,7 +100,7 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         resp = await destination_client.put(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             resource,
         )
 
@@ -109,7 +109,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/logs_custom_pipelines.py

@@ -53,7 +53,7 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         resp = await destination_client.put(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             resource,
         )
 
@@ -62,7 +62,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:

datadog_sync/model/logs_indexes.py

@@ -50,7 +50,7 @@ async def pre_apply_hook(self) -> None:
 
     async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         if _id in self.destination_logs_indexes:
-            self.resource_config.destination_resources[_id] = self.destination_logs_indexes[_id]
+            self.config.state.destination[self.resource_type][_id] = self.destination_logs_indexes[_id]
             return await self.update_resource(_id, resource)
 
         destination_client = self.config.destination_client
@@ -65,20 +65,20 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         # Can't update name so remove it
         resource.pop("name")
         resp = await destination_client.put(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['name']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['name']}",
             resource,
         )
 
-        self.resource_config.destination_resources[_id].update(resp)
-        if not self.resource_config.destination_resources[_id].get("daily_limit"):
-            self.resource_config.destination_resources[_id]["disable_daily_limit"] = True
+        self.config.state.destination[self.resource_type][_id].update(resp)
+        if not self.config.state.destination[self.resource_type][_id].get("daily_limit"):
+            self.config.state.destination[self.resource_type][_id]["disable_daily_limit"] = True
         else:
-            self.resource_config.destination_resources[_id].pop("disable_daily_limit", None)
+            self.config.state.destination[self.resource_type][_id].pop("disable_daily_limit", None)
 
-        return _id, self.resource_config.destination_resources[_id]
+        return _id, self.config.state.destination[self.resource_type][_id]
 
     async def delete_resource(self, _id: str) -> None:
-        index_name = self.resource_config.destination_resources[_id]["name"]
+        index_name = self.config.state.destination[self.resource_type][_id]["name"]
         index_order = await self.config.destination_client.get(self.logs_indexes_order_url)
         if index_name in index_order["index_names"]:
             self.config.logger.warning(

datadog_sync/model/logs_indexes_order.py

@@ -57,11 +57,11 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         if not self.destination_indexes_order:
             raise Exception("Failed to retrieve destination orgs logs index order")
 
-        self.resource_config.destination_resources[_id] = self.destination_indexes_order
+        self.config.state.destination[self.resource_type][_id] = self.destination_indexes_order
         return await self.update_resource(_id, resource)
 
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
-        destination_resources = self.destination_indexes_order or self.resource_config.destination_resources[_id]
+        destination_resources = self.destination_indexes_order or self.config.state.destination[self.resource_type][_id]
         self.handle_additional_indexes(resource, destination_resources)
 
         destination_client = self.config.destination_client
@@ -73,7 +73,7 @@ async def delete_resource(self, _id: str) -> None:
         pass
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
-        logs_indexes = self.config.resources["logs_indexes"].resource_config.destination_resources
+        logs_indexes = self.config.state.destination["logs_indexes"]
 
         failed_connections = []
         for i, name in enumerate(r_obj[key]):

datadog_sync/model/logs_metrics.py

@@ -49,17 +49,17 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         payload = {"data": resource}
         resp = await destination_client.patch(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}",
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
             payload,
         )
 
-        self.resource_config.destination_resources[_id] = resp["data"]
+        self.config.state.destination[self.resource_type][_id] = resp["data"]
         return _id, resp["data"]
 
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
         await destination_client.delete(
-            self.resource_config.base_path + f"/{self.resource_config.destination_resources[_id]['id']}"
+            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]: