[FEATURE] Cloud Persist Assets immediately (#7748)

great_expectations/data_context/data_context/abstract_data_context.py

@@ -370,7 +370,9 @@ def _init_project_config(
     def _init_variables(self) -> DataContextVariables:
         raise NotImplementedError
 
-    def _save_project_config(self) -> None:
+    def _save_project_config(
+        self, _fds_datasource: FluentDatasource | None = None
+    ) -> None:
         """
         Each DataContext will define how its project_config will be saved through its internal 'variables'.
             - FileDataContext : Filesystem.

great_expectations/data_context/data_context/cloud_data_context.py

@@ -58,6 +58,7 @@
         ConfigurationIdentifier,
         ExpectationSuiteIdentifier,
     )
+    from great_expectations.datasource.fluent import Datasource as FluentDatasource
     from great_expectations.render.renderer.site_builder import SiteBuilder
 
 logger = logging.getLogger(__name__)
@@ -820,11 +821,24 @@ def _persist_suite_with_store(
 
         return expectation_suite
 
-    def _save_project_config(self) -> None:
+    def _save_project_config(
+        self, _fds_datasource: FluentDatasource | None = None
+    ) -> None:
         """
         See parent 'AbstractDataContext._save_project_config()` for more information.
 
         Explicitly override base class implementation to retain legacy behavior.
         """
-        logger.debug(f"{type(self).__name__}._save_project_config() is a NoOp")
-        # TODO: this may be adjusted as part of fluent datasources save flow
+        # 042723 kilo59
+        # Currently CloudDataContext and FileDataContext diverge in how FDS are persisted.
+        # FileDataContexts don't use the DatasourceStore at all to save or hydrate FDS configs.
+        # CloudDataContext does use DatasourceStore in order to make use of the Cloud http clients.
+        # The intended future state is for a new FluentDatasourceStore that can fully encapsulate
+        # the different requirements for FDS vs BDS.
+        # At which time `_save_project_config` will revert to being a no-op operation on the CloudDataContext.
+        if _fds_datasource:
+            self._datasource_store.set(key=None, value=_fds_datasource)
+        else:
+            logger.debug(
+                "CloudDataContext._save_project_config() has no `fds_datasource` to update"
+            )

great_expectations/data_context/data_context/file_data_context.py

@@ -128,7 +128,7 @@ def _init_variables(self) -> FileDataContextVariables:
         )
         return variables
 
-    def _save_project_config(self) -> None:
+    def _save_project_config(self, _fds_datasource=None) -> None:
         """
         See parent 'AbstractDataContext._save_project_config()` for more information.
 

great_expectations/data_context/data_context/serializable_data_context.py

@@ -82,7 +82,7 @@ def root_directory(self) -> str:
         return self._context_root_directory
 
     @abc.abstractmethod
-    def _save_project_config(self) -> None:
+    def _save_project_config(self, _fds_datasource=None) -> None:
         """
         See parent 'AbstractDataContext._save_project_config()` for more information.
         Explicitly override base class implementation to retain legacy behavior.

great_expectations/datasource/fluent/interfaces.py

@@ -556,7 +556,7 @@ def _save_context_project_config(self):
         """Check if a DataContext is available and save the project config."""
         if self._data_context:
             try:
-                self._data_context._save_project_config()
+                self._data_context._save_project_config(self)
             except TypeError as type_err:
                 warnings.warn(str(type_err), GxSerializationWarning)
 

tests/datasource/fluent/test_contexts.py

@@ -133,17 +133,14 @@ def test_context_add_or_update_datasource(
         )
         cloud_api_fake.assert_call_count(
             f"{GX_CLOUD_MOCK_BASE_URL}/organizations/{FAKE_ORG_ID}/datasources/{datasource.id}?name={datasource.name}",
-            1,
+            2,
         )
 
         response = requests.get(
             f"{GX_CLOUD_MOCK_BASE_URL}/organizations/{FAKE_ORG_ID}/datasources/{datasource.id}"
         )
         response.raise_for_status()
         print(pf(response.json(), depth=4))
-        pytest.xfail(  # TODO: remove this
-            "Immediate persistance for cloud not implemented"
-        )
         assert response.json()["data"]["attributes"]["datasource_config"].get("assets")
 
 

tests/datasource/fluent/test_metadatasource.py

@@ -100,7 +100,7 @@ def get_datasource(self, datasource_name: str) -> Datasource:
                 f"'{datasource_name}' not found. Available datasources are {list(self._datasources.keys())}"
             ) from exc
 
-    def _save_project_config(self) -> None:
+    def _save_project_config(self, _fs_datasource=None) -> None:
         ...