Cleanup and update typing

datadog_sync/cli.py

@@ -2,7 +2,6 @@
 # under the 3-clause BSD style license (see LICENSE).
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
-
 import sys
 from click import group
 

datadog_sync/commands/shared/options.py

@@ -2,16 +2,21 @@
 # under the 3-clause BSD style license (see LICENSE).
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
+from __future__ import annotations
 import configobj
 from sys import exit
 
 from click import Choice, Option, option, File
 
 from datadog_sync import constants
+from typing import TYPE_CHECKING, Any, Callable, Dict, List
+
+if TYPE_CHECKING:
+    from click.core import Context
 
 
 class CustomOptionClass(Option):
-    def handle_parse_result(self, ctx, opts, args):
+    def handle_parse_result(self, ctx: Context, opts: Dict[Any, Any], args: List[Any]) -> Any:
         try:
             return super(Option, self).handle_parse_result(ctx, opts, args)
         except Exception as e:
@@ -75,7 +80,7 @@ def handle_parse_result(self, ctx, opts, args):
 ]
 
 
-def click_config_file_provider(ctx, opts, value):
+def click_config_file_provider(ctx: Context, opts: CustomOptionClass, value: None) -> None:
     config = configobj.ConfigObj(value, unrepr=True)
     ctx.default_map = ctx.default_map or {}
     ctx.default_map.update(config)
@@ -174,23 +179,23 @@ def click_config_file_provider(ctx, opts, value):
 ]
 
 
-def source_auth_options(func):
+def source_auth_options(func: Callable) -> Callable:
     return _build_options_helper(func, _source_auth_options)
 
 
-def destination_auth_options(func):
+def destination_auth_options(func: Callable) -> Callable:
     return _build_options_helper(func, _destination_auth_options)
 
 
-def common_options(func):
+def common_options(func: Callable) -> Callable:
     return _build_options_helper(func, _common_options)
 
 
-def non_import_common_options(func):
+def non_import_common_options(func: Callable) -> Callable:
     return _build_options_helper(func, _non_import_common_options)
 
 
-def _build_options_helper(func, options):
+def _build_options_helper(func: Callable, options: List[Callable]) -> Callable:
     for _option in options:
         func = _option(func)
     return func

datadog_sync/model/dashboard_lists.py

@@ -3,13 +3,16 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
+from __future__ import annotations
 import copy
-from typing import Optional, List, Dict
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
 from datadog_sync.utils.resource_utils import CustomClientHTTPError, check_diff
 
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
+
 
 class DashboardLists(BaseResource):
     resource_type = "dashboard_lists"
@@ -19,7 +22,7 @@ class DashboardLists(BaseResource):
         excluded_attributes=["id", "type", "author", "created", "modified", "is_favorite", "dashboard_count"],
     )
     # Additional Dashboards specific attributes
-    dash_list_items_path = "/api/v2/dashboard/lists/manual/{}/dashboards"
+    dash_list_items_path: str = "/api/v2/dashboard/lists/manual/{}/dashboards"
 
     def get_resources(self, client: CustomClient) -> List[Dict]:
         resp = client.get(self.resource_config.base_path).json()
@@ -32,6 +35,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
         if _id:
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()
 
+        resource = cast(dict, resource)
         _id = str(resource["id"])
         resp = None
         try:

datadog_sync/model/dashboards.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class Dashboards(BaseResource):
@@ -31,8 +34,9 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
         source_client = self.config.source_client
         import_id = _id or resource["id"]
 
-        dashboard = source_client.get(self.resource_config.base_path + f"/{import_id}").json()
-        self.resource_config.source_resources[import_id] = dashboard
+        resource = source_client.get(self.resource_config.base_path + f"/{import_id}").json()
+        resource = cast(dict, resource)
+        self.resource_config.source_resources[import_id] = resource
 
     def pre_resource_action_hook(self, _id, resource: Dict) -> None:
         pass

datadog_sync/model/downtimes.py

@@ -2,12 +2,15 @@
 # under the 3-clause BSD style license (see LICENSE).
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
+from __future__ import annotations
 import math
-from typing import Optional, List, Dict
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 from datetime import datetime
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 RECURRING_TIMES = {
@@ -48,6 +51,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()
 
+        resource = cast(dict, resource)
         if resource["canceled"]:
             return
         # Dispose the recurring child downtimes and only retain the parent

datadog_sync/model/host_tags.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class HostTags(BaseResource):
@@ -24,6 +27,8 @@ def get_resources(self, client: CustomClient) -> List[Dict]:
     def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] = None) -> None:
         if _id:
             return  # This should never occur. No resource depends on it.
+
+        resource = cast(dict, resource)
         tag = resource[0]
         hosts = resource[1]
         for host in hosts:

datadog_sync/model/logs_custom_pipelines.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class LogsCustomPipelines(BaseResource):
@@ -28,6 +31,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()
 
+        resource = cast(dict, resource)
         if resource["is_read_only"]:
             return
         self.resource_config.source_resources[resource["id"]] = resource

datadog_sync/model/logs_indexes.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class LogsIndexes(BaseResource):
@@ -31,6 +34,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()
 
+        resource = cast(dict, resource)
         if not resource.get("daily_limit"):
             resource["disable_daily_limit"] = True
         self.resource_config.source_resources[resource["name"]] = resource

datadog_sync/model/logs_metrics.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class LogsMetrics(BaseResource):
@@ -26,6 +29,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()["data"]
 
+        resource = cast(dict, resource)
         self.resource_config.source_resources[resource["id"]] = resource
 
     def pre_resource_action_hook(self, _id, resource: Dict) -> None:

datadog_sync/model/logs_restriction_queries.py

@@ -3,12 +3,16 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict, Tuple
+from __future__ import annotations
+from typing import TYPE_CHECKING, Any, Optional, List, Dict, Tuple, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient, PaginationConfig
+from datadog_sync.utils.custom_client import PaginationConfig
 from datadog_sync.utils.resource_utils import CustomClientHTTPError, check_diff
 
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
+
 
 class LogsRestrictionQueries(BaseResource):
     resource_type = "logs_restriction_queries"
@@ -35,7 +39,7 @@ def get_resources(self, client: CustomClient) -> List[Dict]:
         )
         return resp
 
-    def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] = None) -> None:
+    def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict[str, Any]] = None) -> None:
         source_client = self.config.source_client
         import_id = _id or resource["id"]
 

datadog_sync/model/metric_tag_configurations.py

@@ -3,10 +3,13 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class MetricTagConfigurations(BaseResource):
@@ -28,6 +31,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}/tags").json()["data"]
 
+        resource = cast(dict, resource)
         self.resource_config.source_resources[resource["id"]] = resource
 
     def pre_resource_action_hook(self, _id, resource: Dict) -> None:

datadog_sync/model/monitors.py

@@ -3,12 +3,14 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
+from __future__ import annotations
 import re
-from typing import Optional, List, Dict
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient
-from datadog_sync.utils.resource_utils import ResourceConnectionError
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class Monitors(BaseResource):
@@ -41,6 +43,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()
 
+        resource = cast(dict, resource)
         if resource["type"] in ("synthetics alert", "slo alert"):
             return
 
@@ -97,6 +100,8 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
                     failed_connections.append(_id)
             r_obj[key] = (r_obj[key].replace("#", "")).strip()
             return failed_connections
-        elif key != "query":
+        elif key == "query":
+            return None
+        else:
             # Use default connect_id method in base class when not handling special case for `query`
             return super(Monitors, self).connect_id(key, r_obj, resource_to_connect)

datadog_sync/model/notebooks.py

@@ -3,10 +3,14 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 
-from typing import Optional, List, Dict
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, List, Dict, cast
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.custom_client import CustomClient, PaginationConfig
+from datadog_sync.utils.custom_client import PaginationConfig
+
+if TYPE_CHECKING:
+    from datadog_sync.utils.custom_client import CustomClient
 
 
 class Notebooks(BaseResource):
@@ -42,6 +46,7 @@ def import_resource(self, _id: Optional[str] = None, resource: Optional[Dict] =
             source_client = self.config.source_client
             resource = source_client.get(self.resource_config.base_path + f"/{_id}").json()["data"]
 
+        resource = cast(dict, resource)
         self.handle_special_case_attr(resource)
         self.resource_config.source_resources[resource["id"]] = resource