dev: immutable args remove (datahub-project#12735)

metadata-ingestion/src/datahub/cli/lite_cli.py

@@ -285,10 +285,12 @@ def search(
     ctx: click.Context,
     query: str = "",
     flavor: str = SearchFlavor.FREE_TEXT.name.lower(),
-    aspect: List[str] = [],
+    aspect: Optional[List[str]] = None,
     details: bool = True,
 ) -> None:
     """Search with a free text or exact query string"""
+    if aspect is None:
+        aspect = []
 
     # query flavor should be sanitized by now, but we still need to convert it to a SearchFlavor
     try:

metadata-ingestion/src/datahub/ingestion/graph/client.py

@@ -1547,7 +1547,7 @@ def _run_assertion_result_shared(self) -> str:
         return fragment
 
     def _run_assertion_build_params(
-        self, params: Optional[Dict[str, str]] = {}
+        self, params: Optional[Dict[str, str]] = None
     ) -> List[Any]:
         if params is None:
             return []
@@ -1566,9 +1566,11 @@ def run_assertion(
         self,
         urn: str,
         save_result: bool = True,
-        parameters: Optional[Dict[str, str]] = {},
+        parameters: Optional[Dict[str, str]] = None,
         async_flag: bool = False,
     ) -> Dict:
+        if parameters is None:
+            parameters = {}
         params = self._run_assertion_build_params(parameters)
         graph_query: str = """
             %s
@@ -1597,9 +1599,11 @@ def run_assertions(
         self,
         urns: List[str],
         save_result: bool = True,
-        parameters: Optional[Dict[str, str]] = {},
+        parameters: Optional[Dict[str, str]] = None,
         async_flag: bool = False,
     ) -> Dict:
+        if parameters is None:
+            parameters = {}
         params = self._run_assertion_build_params(parameters)
         graph_query: str = """
             %s
@@ -1636,10 +1640,14 @@ def run_assertions(
     def run_assertions_for_asset(
         self,
         urn: str,
-        tag_urns: Optional[List[str]] = [],
-        parameters: Optional[Dict[str, str]] = {},
+        tag_urns: Optional[List[str]] = None,
+        parameters: Optional[Dict[str, str]] = None,
         async_flag: bool = False,
     ) -> Dict:
+        if tag_urns is None:
+            tag_urns = []
+        if parameters is None:
+            parameters = {}
         params = self._run_assertion_build_params(parameters)
         graph_query: str = """
             %s

metadata-ingestion/tests/integration/powerbi/test_admin_only_api.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from unittest import mock
 
 import pytest
@@ -49,7 +49,11 @@ def admin_datasets_response(request, context):
         }
 
 
-def register_mock_admin_api(request_mock: Any, override_data: dict = {}) -> None:
+def register_mock_admin_api(
+    request_mock: Any, override_data: Optional[dict] = None
+) -> None:
+    if override_data is None:
+        override_data = {}
     api_vs_response = {
         "https://api.powerbi.com/v1.0/myorg/admin/groups/64ED5CAD-7C10-4684-8180-826122881108/datasets": {
             "method": "GET",

metadata-ingestion/tests/integration/powerbi/test_m_parser.py

@@ -1,7 +1,7 @@
 import logging
 import sys
 import time
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -93,10 +93,12 @@ def get_data_platform_tables_with_dummy_table(
 
 
 def get_default_instances(
-    override_config: dict = {},
+    override_config: Optional[dict] = None,
 ) -> Tuple[
     PipelineContext, PowerBiDashboardSourceConfig, AbstractDataPlatformInstanceResolver
 ]:
+    if override_config is None:
+        override_config = {}
     config: PowerBiDashboardSourceConfig = PowerBiDashboardSourceConfig.parse_obj(
         {
             "tenant_id": "fake",

metadata-ingestion/tests/integration/powerbi/test_powerbi.py

@@ -1463,8 +1463,10 @@ def common_app_ingest(
     pytestconfig: pytest.Config,
     requests_mock: Any,
     output_mcp_path: str,
-    override_config: dict = {},
+    override_config: Optional[dict] = None,
 ) -> Pipeline:
+    if override_config is None:
+        override_config = {}
     register_mock_api(
         pytestconfig=pytestconfig,
         request_mock=requests_mock,

metadata-ingestion/tests/integration/powerbi/test_profiling.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from unittest import mock
 
 from freezegun import freeze_time
@@ -103,7 +103,11 @@ def execute_queries_response(request, context):
         }
 
 
-def register_mock_admin_api(request_mock: Any, override_data: dict = {}) -> None:
+def register_mock_admin_api(
+    request_mock: Any, override_data: Optional[dict] = None
+) -> None:
+    if override_data is None:
+        override_data = {}
     api_vs_response = {
         "https://api.powerbi.com/v1.0/myorg/groups/64ED5CAD-7C10-4684-8180-826122881108/datasets": {
             "method": "GET",

metadata-ingestion/tests/integration/powerbi_report_server/test_powerbi_report_server.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Any, Optional
 from unittest import mock
 
 from freezegun import freeze_time
@@ -21,7 +22,11 @@ def mock_user_to_add(*args, **kwargs):
     return None
 
 
-def register_mock_api(request_mock, override_mock_data={}):
+def register_mock_api(
+    request_mock: Any, override_mock_data: Optional[dict] = None
+) -> None:
+    if override_mock_data is None:
+        override_mock_data = {}
     api_vs_response = {
         "https://host_port/Reports/api/v2.0/Reports": {
             "method": "GET",

metadata-ingestion/tests/integration/qlik_sense/test_qlik_sense.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from unittest.mock import patch
 
 import pytest
@@ -7,7 +7,9 @@
 from tests.test_helpers import mce_helpers
 
 
-def register_mock_api(request_mock: Any, override_data: dict = {}) -> None:
+def register_mock_api(request_mock: Any, override_data: Optional[dict] = None) -> None:
+    if override_data is None:
+        override_data = {}
     api_vs_response: Dict[str, Dict] = {
         "https://iq37k6byr9lgam8.us.qlikcloud.com/api/v1/api-keys": {
             "method": "GET",

metadata-ingestion/tests/integration/sigma/test_sigma.py

@@ -1,12 +1,14 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 import pytest
 
 from datahub.ingestion.run.pipeline import Pipeline
 from tests.test_helpers import mce_helpers
 
 
-def register_mock_api(request_mock: Any, override_data: dict = {}) -> None:
+def register_mock_api(request_mock: Any, override_data: Optional[dict] = None) -> None:
+    if override_data is None:
+        override_data = {}
     api_vs_response: Dict[str, Dict] = {
         "https://aws-api.sigmacomputing.com/v2/auth/token": {
             "method": "POST",

metadata-ingestion/tests/integration/superset/test_superset.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from unittest.mock import patch
 
 import pytest
@@ -17,7 +17,9 @@
 GMS_SERVER = f"http://localhost:{GMS_PORT}"
 
 
-def register_mock_api(request_mock: Any, override_data: dict = {}) -> None:
+def register_mock_api(request_mock: Any, override_data: Optional[dict] = None) -> None:
+    if override_data is None:
+        override_data = {}
     api_vs_response = {
         "mock://mock-domain.superset.com/api/v1/security/login": {
             "method": "POST",

metadata-ingestion/tests/test_helpers/mce_helpers.py

@@ -262,9 +262,11 @@ def assert_for_each_entity(
     aspect_name: str,
     aspect_field_matcher: Dict[str, Any],
     file: str,
-    exception_urns: List[str] = [],
+    exception_urns: Optional[List[str]] = None,
 ) -> int:
     """Assert that an aspect name with the desired fields exists for each entity urn"""
+    if exception_urns is None:
+        exception_urns = []
     test_output = load_json_file(file)
     assert isinstance(test_output, list)
     # mce urns

metadata-ingestion/tests/test_helpers/test_connection_helpers.py

@@ -33,9 +33,13 @@ def assert_basic_connectivity_failure(
 
 def assert_capability_report(
     capability_report: Optional[Dict[Union[SourceCapability, str], CapabilityReport]],
-    success_capabilities: List[SourceCapability] = [],
-    failure_capabilities: Dict[SourceCapability, str] = {},
+    success_capabilities: Optional[List[SourceCapability]] = None,
+    failure_capabilities: Optional[Dict[SourceCapability, str]] = None,
 ) -> None:
+    if success_capabilities is None:
+        success_capabilities = []
+    if failure_capabilities is None:
+        failure_capabilities = {}
     assert capability_report
     for capability in success_capabilities:
         assert capability_report[capability]

metadata-ingestion/tests/unit/api/source_helpers/test_auto_browse_path_v2.py

@@ -1,5 +1,5 @@
 from itertools import zip_longest
-from typing import Any, Dict, Iterable, List
+from typing import Any, Dict, Iterable, List, Optional
 from unittest.mock import patch
 
 import datahub.metadata.schema_classes as models
@@ -403,9 +403,11 @@ def _get_browse_paths_from_wu(
 
 def _create_container_aspects(
     d: Dict[str, Any],
-    other_aspects: Dict[str, List[models._Aspect]] = {},
+    other_aspects: Optional[Dict[str, List[models._Aspect]]] = None,
     root: bool = True,
 ) -> Iterable[MetadataWorkUnit]:
+    if other_aspects is None:
+        other_aspects = {}
     for k, v in d.items():
         urn = make_container_urn(k)
         yield MetadataChangeProposalWrapper(