Repository: great-expectations/great_expectations
Pull request #7036 (2023-02-02 16:32:46)

[MAINTENANCE] Add `id` as a param to any CRUD methods with `ge_cloud_id`
Changes proposed in this pull request:
- Effort to remove the notion of `ge_cloud_id` from our `AbstractDataContext`
    - Going to have them exist concurrently and then move to deprecation at a future date



### Definition of Done
- [x] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [x] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_checklist) of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [x] I have run any local integration tests and made sure that nothing is broken.


codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=7036&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations">Click on the image</a> for a new way to code review</b>
</summary>

  - Make big changes easier — review code in small groups of related files
  - Know where to start — see the whole change at a glance
  - Take a code tour — explore the change with an interactive tour
  - Make comments and review — all fully sync’ed with github

    [Try it now!](https://app.codesee.io/r/reviews?pr=7036&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)
</details>

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/great-expectations/great_expectations/7036/109d4842/a6bbdeca866a2501db6740f231e97a48b4ba919c.svg)](https://app.codesee.io/r/reviews?pr=7036&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)

Patch:
diff --git a/great_expectations/data_context/data_context/abstract_data_context.py b/great_expectations/data_context/data_context/abstract_data_context.py
index 18e882432..19580a8f9 100644
--- a/great_expectations/data_context/data_context/abstract_data_context.py
+++ b/great_expectations/data_context/data_context/abstract_data_context.py
@@ -1433,29 +1433,31 @@ class AbstractDataContext(ConfigPeer, ABC):
     def add_checkpoint(
         self,
         name: str,
-        config_version: Optional[Union[int, float]] = None,
-        template_name: Optional[str] = None,
-        module_name: Optional[str] = None,
-        class_name: Optional[str] = None,
-        run_name_template: Optional[str] = None,
-        expectation_suite_name: Optional[str] = None,
-        batch_request: Optional[dict] = None,
-        action_list: Optional[List[dict]] = None,
-        evaluation_parameters: Optional[dict] = None,
-        runtime_configuration: Optional[dict] = None,
-        validations: Optional[List[dict]] = None,
-        profilers: Optional[List[dict]] = None,
+        config_version: int | float | None = None,
+        template_name: str | None = None,
+        module_name: str | None = None,
+        class_name: str | None = None,
+        run_name_template: str | None = None,
+        expectation_suite_name: str | None = None,
+        batch_request: dict | None = None,
+        action_list: list[dict] | None = None,
+        evaluation_parameters: dict | None = None,
+        runtime_configuration: dict | None = None,
+        validations: list[dict] | None = None,
+        profilers: list[dict] | None = None,
         # Next two fields are for LegacyCheckpoint configuration
-        validation_operator_name: Optional[str] = None,
-        batches: Optional[List[dict]] = None,
+        validation_operator_name: str | None = None,
+        batches: list[dict] | None = None,
         # the following four arguments are used by SimpleCheckpoint
-        site_names: Optional[Union[str, List[str]]] = None,
-        slack_webhook: Optional[str] = None,
-        notify_on: Optional[str] = None,
-        notify_with: Optional[Union[str, List[str]]] = None,
-        ge_cloud_id: Optional[str] = None,
-        expectation_suite_ge_cloud_id: Optional[str] = None,
-        default_validation_id: Optional[str] = None,
+        site_names: str | list[str] | None = None,
+        slack_webhook: str | None = None,
+        notify_on: str | None = None,
+        notify_with: str | list[str] | None = None,
+        ge_cloud_id: str | None = None,
+        expectation_suite_ge_cloud_id: str | None = None,
+        default_validation_id: str | None = None,
+        id: str | None = None,
+        expectation_suite_id: str | None = None,
     ) -> Checkpoint:
         """Add a Checkpoint to the DataContext.
 
@@ -1489,6 +1491,14 @@ class AbstractDataContext(ConfigPeer, ABC):
         Returns:
             The Checkpoint object created.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        expectation_suite_id = self._resolve_id_and_ge_cloud_id(
+            id=expectation_suite_id, ge_cloud_id=expectation_suite_ge_cloud_id
+        )
+        del ge_cloud_id
+        del expectation_suite_ge_cloud_id
+
         from great_expectations.checkpoint.checkpoint import Checkpoint
 
         checkpoint: Checkpoint = Checkpoint.construct_from_config_args(
@@ -1515,18 +1525,19 @@ class AbstractDataContext(ConfigPeer, ABC):
             slack_webhook=slack_webhook,
             notify_on=notify_on,
             notify_with=notify_with,
-            ge_cloud_id=ge_cloud_id,
-            expectation_suite_ge_cloud_id=expectation_suite_ge_cloud_id,
+            ge_cloud_id=id,
+            expectation_suite_ge_cloud_id=expectation_suite_id,
             default_validation_id=default_validation_id,
         )
 
-        self.checkpoint_store.add_checkpoint(checkpoint, name, ge_cloud_id)
+        self.checkpoint_store.add_checkpoint(checkpoint=checkpoint, name=name, id=id)
         return checkpoint
 
     def get_checkpoint(
         self,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> Checkpoint:
         """Retrieves a given Checkpoint by either name or id.
 
@@ -1540,10 +1551,14 @@ class AbstractDataContext(ConfigPeer, ABC):
         Raises:
             CheckpointNotFoundError if the requested Checkpoint does not exists.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         from great_expectations.checkpoint.checkpoint import Checkpoint
 
         checkpoint_config: CheckpointConfig = self.checkpoint_store.get_checkpoint(
-            name=name, ge_cloud_id=ge_cloud_id
+            name=name, id=id
         )
         checkpoint: Checkpoint = Checkpoint.instantiate_from_config_with_runtime_args(
             checkpoint_config=checkpoint_config,
@@ -1556,8 +1571,9 @@ class AbstractDataContext(ConfigPeer, ABC):
     @public_api
     def delete_checkpoint(
         self,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> None:
         """Deletes a given Checkpoint by either name or id.
 
@@ -1568,9 +1584,11 @@ class AbstractDataContext(ConfigPeer, ABC):
         Raises:
             CheckpointNotFoundError if the requested Checkpoint does not exists.
         """
-        return self.checkpoint_store.delete_checkpoint(
-            name=name, ge_cloud_id=ge_cloud_id
-        )
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
+        return self.checkpoint_store.delete_checkpoint(name=name, id=id)
 
     @public_api
     @usage_statistics_enabled_method(
@@ -1578,22 +1596,24 @@ class AbstractDataContext(ConfigPeer, ABC):
     )
     def run_checkpoint(
         self,
-        checkpoint_name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
-        template_name: Optional[str] = None,
-        run_name_template: Optional[str] = None,
-        expectation_suite_name: Optional[str] = None,
-        batch_request: Optional[BatchRequestBase] = None,
-        action_list: Optional[List[dict]] = None,
-        evaluation_parameters: Optional[dict] = None,
-        runtime_configuration: Optional[dict] = None,
-        validations: Optional[List[dict]] = None,
-        profilers: Optional[List[dict]] = None,
-        run_id: Optional[Union[str, int, float]] = None,
-        run_name: Optional[str] = None,
-        run_time: Optional[datetime.datetime] = None,
-        result_format: Optional[str] = None,
-        expectation_suite_ge_cloud_id: Optional[str] = None,
+        checkpoint_name: str | None = None,
+        ge_cloud_id: str | None = None,
+        template_name: str | None = None,
+        run_name_template: str | None = None,
+        expectation_suite_name: str | None = None,
+        batch_request: BatchRequestBase | None = None,
+        action_list: list[dict] | None = None,
+        evaluation_parameters: dict | None = None,
+        runtime_configuration: dict | None = None,
+        validations: list[dict] | None = None,
+        profilers: list[dict] | None = None,
+        run_id: str | int | float | None = None,
+        run_name: str | None = None,
+        run_time: datetime.datetime | None = None,
+        result_format: str | None = None,
+        expectation_suite_ge_cloud_id: str | None = None,
+        id: str | None = None,
+        expectation_suite_id: str | None = None,
         **kwargs,
     ) -> CheckpointResult:
         """Validate using an existing Checkpoint.
@@ -1620,9 +1640,17 @@ class AbstractDataContext(ConfigPeer, ABC):
         Returns:
             CheckpointResult
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        expectation_suite_id = self._resolve_id_and_ge_cloud_id(
+            id=expectation_suite_id, ge_cloud_id=expectation_suite_ge_cloud_id
+        )
+        del ge_cloud_id
+        del expectation_suite_ge_cloud_id
+
         return self._run_checkpoint(
             checkpoint_name=checkpoint_name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
             template_name=template_name,
             run_name_template=run_name_template,
             expectation_suite_name=expectation_suite_name,
@@ -1636,33 +1664,33 @@ class AbstractDataContext(ConfigPeer, ABC):
             run_name=run_name,
             run_time=run_time,
             result_format=result_format,
-            expectation_suite_ge_cloud_id=expectation_suite_ge_cloud_id,
+            expectation_suite_ge_cloud_id=expectation_suite_id,
             **kwargs,
         )
 
     def _run_checkpoint(
         self,
-        checkpoint_name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
-        template_name: Optional[str] = None,
-        run_name_template: Optional[str] = None,
-        expectation_suite_name: Optional[str] = None,
-        batch_request: Optional[BatchRequestBase] = None,
-        action_list: Optional[List[dict]] = None,
-        evaluation_parameters: Optional[dict] = None,
-        runtime_configuration: Optional[dict] = None,
-        validations: Optional[List[dict]] = None,
-        profilers: Optional[List[dict]] = None,
-        run_id: Optional[Union[str, int, float]] = None,
-        run_name: Optional[str] = None,
-        run_time: Optional[datetime.datetime] = None,
-        result_format: Optional[str] = None,
-        expectation_suite_ge_cloud_id: Optional[str] = None,
+        checkpoint_name: str | None = None,
+        id: str | None = None,
+        template_name: str | None = None,
+        run_name_template: str | None = None,
+        expectation_suite_name: str | None = None,
+        batch_request: BatchRequestBase | None = None,
+        action_list: list[dict] | None = None,
+        evaluation_parameters: dict | None = None,
+        runtime_configuration: dict | None = None,
+        validations: list[dict] | None = None,
+        profilers: list[dict] | None = None,
+        run_id: str | int | float | None = None,
+        run_name: str | None = None,
+        run_time: datetime.datetime | None = None,
+        result_format: str | None = None,
+        expectation_suite_ge_cloud_id: str | None = None,
         **kwargs,
     ) -> CheckpointResult:
         checkpoint: Checkpoint = self.get_checkpoint(
             name=checkpoint_name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
         result: CheckpointResult = checkpoint.run_with_runtime_args(
             template_name=template_name,
@@ -1756,7 +1784,8 @@ class AbstractDataContext(ConfigPeer, ABC):
         expectation_suite: Optional[ExpectationSuite] = None,
         create_expectation_suite_with_name: Optional[str] = None,
         include_rendered_content: Optional[bool] = None,
-        **kwargs: Optional[dict],
+        expectation_suite_id: Optional[str] = None,
+        **kwargs,
     ) -> Validator:
         """Retrieve a Validator with a batch list and an `ExpectationSuite`.
 
@@ -1812,6 +1841,12 @@ class AbstractDataContext(ConfigPeer, ABC):
                 of `batch_data`, `query` or `path`), or if the `ExpectationSuite` cannot be created or
                 retrieved using either the provided name or identifier
         """
+        # <GX_RENAME>
+        expectation_suite_id = self._resolve_id_and_ge_cloud_id(
+            id=expectation_suite_id, ge_cloud_id=expectation_suite_ge_cloud_id
+        )
+        del expectation_suite_ge_cloud_id
+
         include_rendered_content = (
             self._determine_if_expectation_validation_result_include_rendered_content(
                 include_rendered_content=include_rendered_content
@@ -1825,7 +1860,7 @@ class AbstractDataContext(ConfigPeer, ABC):
                     expectation_suite is not None,
                     expectation_suite_name is not None,
                     create_expectation_suite_with_name is not None,
-                    expectation_suite_ge_cloud_id is not None,
+                    expectation_suite_id is not None,
                 ]
             )
             > 1
@@ -1835,14 +1870,14 @@ class AbstractDataContext(ConfigPeer, ABC):
             )
             raise ValueError(
                 "No more than one of expectation_suite_name,"
-                f"{'expectation_suite_ge_cloud_id,' if ge_cloud_mode else ''}"
+                f"{'expectation_suite_id,' if ge_cloud_mode else ''}"
                 " expectation_suite, or create_expectation_suite_with_name can be specified"
             )
 
-        if expectation_suite_ge_cloud_id is not None:
+        if expectation_suite_id is not None:
             expectation_suite = self.get_expectation_suite(
                 include_rendered_content=include_rendered_content,
-                ge_cloud_id=expectation_suite_ge_cloud_id,
+                ge_cloud_id=expectation_suite_id,
             )
         if expectation_suite_name is not None:
             expectation_suite = self.get_expectation_suite(
@@ -2338,8 +2373,9 @@ class AbstractDataContext(ConfigPeer, ABC):
 
     def delete_expectation_suite(
         self,
-        expectation_suite_name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        expectation_suite_name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> bool:
         """Delete specified expectation suite from data_context expectation store.
 
@@ -2361,9 +2397,9 @@ class AbstractDataContext(ConfigPeer, ABC):
     @deprecated_argument(argument_name="ge_cloud_id", version="0.15.45")
     def get_expectation_suite(
         self,
-        expectation_suite_name: Optional[str] = None,
-        include_rendered_content: Optional[bool] = None,
-        ge_cloud_id: Optional[str] = None,
+        expectation_suite_name: str | None = None,
+        include_rendered_content: bool | None = None,
+        ge_cloud_id: str | None = None,
     ) -> ExpectationSuite:
         """Get an Expectation Suite by name.
 
@@ -2474,8 +2510,9 @@ class AbstractDataContext(ConfigPeer, ABC):
 
     def get_profiler(
         self,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> RuleBasedProfiler:
         """Retrieves a given Profiler by either name or id.
 
@@ -2489,17 +2526,22 @@ class AbstractDataContext(ConfigPeer, ABC):
         Raises:
             ProfilerNotFoundError if the requested Profiler does not exists.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         return RuleBasedProfiler.get_profiler(
             data_context=self,
             profiler_store=self.profiler_store,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
     def delete_profiler(
         self,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> None:
         """Deletes a given Profiler by either name or id.
 
@@ -2510,10 +2552,14 @@ class AbstractDataContext(ConfigPeer, ABC):
         Raises:
             ProfilerNotFoundError if the requested Profiler does not exists.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         RuleBasedProfiler.delete_profiler(
             profiler_store=self.profiler_store,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
     def update_profiler(self, profiler: RuleBasedProfiler) -> RuleBasedProfiler:
@@ -2566,12 +2612,13 @@ class AbstractDataContext(ConfigPeer, ABC):
     )
     def run_profiler_with_dynamic_arguments(
         self,
-        batch_list: Optional[List[Batch]] = None,
-        batch_request: Optional[Union[BatchRequestBase, dict]] = None,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
-        variables: Optional[dict] = None,
-        rules: Optional[dict] = None,
+        batch_list: list[Batch] | None = None,
+        batch_request: BatchRequestBase | dict | None = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        variables: dict | None = None,
+        rules: dict | None = None,
+        id: str | None = None,
     ) -> RuleBasedProfilerResult:
         """Retrieve a RuleBasedProfiler from a ProfilerStore and run it with rules/variables supplied at runtime.
 
@@ -2590,11 +2637,15 @@ class AbstractDataContext(ConfigPeer, ABC):
             AssertionError if both a `name` and `ge_cloud_id` are provided.
             AssertionError if both an `expectation_suite` and `expectation_suite_name` are provided.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         return self._run_profiler_with_dynamic_arguments(
             batch_list=batch_list,
             batch_request=batch_request,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
             variables=variables,
             rules=rules,
         )
@@ -2604,7 +2655,7 @@ class AbstractDataContext(ConfigPeer, ABC):
         batch_list: Optional[List[Batch]] = None,
         batch_request: Optional[Union[BatchRequestBase, dict]] = None,
         name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        id: Optional[str] = None,
         variables: Optional[dict] = None,
         rules: Optional[dict] = None,
     ) -> RuleBasedProfilerResult:
@@ -2614,7 +2665,7 @@ class AbstractDataContext(ConfigPeer, ABC):
             batch_list=batch_list,
             batch_request=batch_request,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
             variables=variables,
             rules=rules,
         )
@@ -2624,10 +2675,11 @@ class AbstractDataContext(ConfigPeer, ABC):
     )
     def run_profiler_on_data(
         self,
-        batch_list: Optional[List[Batch]] = None,
-        batch_request: Optional[BatchRequestBase] = None,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        batch_list: list[Batch] | None = None,
+        batch_request: BatchRequestBase | None = None,
+        name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> RuleBasedProfilerResult:
         """Retrieve a RuleBasedProfiler from a ProfilerStore and run it with a batch request supplied at runtime.
 
@@ -2645,19 +2697,23 @@ class AbstractDataContext(ConfigPeer, ABC):
             AssertionError if both a `name` and `ge_cloud_id` are provided.
             AssertionError if both an `expectation_suite` and `expectation_suite_name` are provided.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         return self._run_profiler_on_data(
             batch_list=batch_list,
             batch_request=batch_request,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
     def _run_profiler_on_data(
         self,
-        batch_list: Optional[List[Batch]] = None,
-        batch_request: Optional[BatchRequestBase] = None,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        batch_list: list[Batch] | None = None,
+        batch_request: BatchRequestBase | None = None,
+        name: str | None = None,
+        id: str | None = None,
     ) -> RuleBasedProfilerResult:
         return RuleBasedProfiler.run_profiler_on_data(
             data_context=self,
@@ -2665,7 +2721,7 @@ class AbstractDataContext(ConfigPeer, ABC):
             batch_list=batch_list,
             batch_request=batch_request,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
     def add_validation_operator(
@@ -4674,3 +4730,11 @@ Generated, evaluated, and stored {total_expectations} Expectations during profil
         if xdatasources:
             self.zep_config.xdatasources.update(xdatasources)
         return self.zep_config.xdatasources
+
+    @staticmethod
+    def _resolve_id_and_ge_cloud_id(
+        id: str | None, ge_cloud_id: str | None
+    ) -> str | None:
+        if id and ge_cloud_id:
+            raise ValueError("Please only pass in either id or ge_cloud_id (not both)")
+        return id or ge_cloud_id
diff --git a/great_expectations/data_context/data_context/cloud_data_context.py b/great_expectations/data_context/data_context/cloud_data_context.py
index f030aab25..18173ea5f 100644
--- a/great_expectations/data_context/data_context/cloud_data_context.py
+++ b/great_expectations/data_context/data_context/cloud_data_context.py
@@ -506,8 +506,9 @@ class CloudDataContext(SerializableDataContext):
 
     def delete_expectation_suite(
         self,
-        expectation_suite_name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        expectation_suite_name: str | None = None,
+        ge_cloud_id: str | None = None,
+        id: str | None = None,
     ) -> bool:
         """Delete specified expectation suite from data_context expectation store.
 
@@ -517,9 +518,13 @@ class CloudDataContext(SerializableDataContext):
         Returns:
             True for Success and False for Failure.
         """
+        # <GX_RENAME>
+        id = self._resolve_id_and_ge_cloud_id(id=id, ge_cloud_id=ge_cloud_id)
+        del ge_cloud_id
+
         key = GXCloudIdentifier(
             resource_type=GXCloudRESTResource.EXPECTATION_SUITE,
-            cloud_id=ge_cloud_id,
+            cloud_id=id,
         )
 
         return self.expectations_store.remove_key(key)
diff --git a/great_expectations/data_context/migrator/configuration_bundle.py b/great_expectations/data_context/migrator/configuration_bundle.py
index 151393f2a..4691d104a 100644
--- a/great_expectations/data_context/migrator/configuration_bundle.py
+++ b/great_expectations/data_context/migrator/configuration_bundle.py
@@ -119,9 +119,7 @@ class ConfigurationBundle:
 
     def _get_all_checkpoints(self) -> List[CheckpointConfig]:
         return [
-            self._context.checkpoint_store.get_checkpoint(
-                name=checkpoint_name, ge_cloud_id=None
-            )
+            self._context.checkpoint_store.get_checkpoint(name=checkpoint_name, id=None)
             for checkpoint_name in self._context.list_checkpoints()
         ]
 
diff --git a/great_expectations/data_context/store/checkpoint_store.py b/great_expectations/data_context/store/checkpoint_store.py
index 0ed7b1e20..dc6ff5256 100644
--- a/great_expectations/data_context/store/checkpoint_store.py
+++ b/great_expectations/data_context/store/checkpoint_store.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import itertools
 import logging
 import os
@@ -116,11 +118,11 @@ class CheckpointStore(ConfigurationStore):
 
     def delete_checkpoint(
         self,
-        name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        name: str | None = None,
+        id: str | None = None,
     ) -> None:
         key: Union[GXCloudIdentifier, ConfigurationIdentifier] = self.determine_key(
-            name=name, ge_cloud_id=ge_cloud_id
+            name=name, ge_cloud_id=id
         )
         try:
             self.remove_key(key=key)
@@ -129,11 +131,9 @@ class CheckpointStore(ConfigurationStore):
                 message=f'Non-existent Checkpoint configuration named "{key.configuration_key}".\n\nDetails: {exc_ik}'  # type: ignore[union-attr]
             )
 
-    def get_checkpoint(
-        self, name: Optional[str], ge_cloud_id: Optional[str]
-    ) -> CheckpointConfig:
+    def get_checkpoint(self, name: str | None, id: str | None) -> CheckpointConfig:
         key: Union[GXCloudIdentifier, ConfigurationIdentifier] = self.determine_key(
-            name=name, ge_cloud_id=ge_cloud_id
+            name=name, ge_cloud_id=id
         )
         try:
             checkpoint_config: CheckpointConfig = self.get(key=key)  # type: ignore[assignment]
@@ -169,16 +169,16 @@ class CheckpointStore(ConfigurationStore):
         return checkpoint_config
 
     def add_checkpoint(
-        self, checkpoint: "Checkpoint", name: Optional[str], ge_cloud_id: Optional[str]
+        self, checkpoint: Checkpoint, name: str | None, id: str | None
     ) -> None:
         key: Union[GXCloudIdentifier, ConfigurationIdentifier] = self.determine_key(
-            name=name, ge_cloud_id=ge_cloud_id
+            name=name, ge_cloud_id=id
         )
         checkpoint_config: CheckpointConfig = checkpoint.get_config()  # type: ignore[assignment]
         checkpoint_ref = self.set(key=key, value=checkpoint_config)  # type: ignore[func-returns-value]
         if isinstance(checkpoint_ref, GXCloudIDAwareRef):
-            ge_cloud_id = checkpoint_ref.cloud_id
-            checkpoint.ge_cloud_id = uuid.UUID(ge_cloud_id)  # type: ignore[misc]
+            cloud_id = checkpoint_ref.cloud_id
+            checkpoint.ge_cloud_id = uuid.UUID(cloud_id)  # type: ignore[misc]
 
     def create(self, checkpoint_config: CheckpointConfig) -> Optional[DataContextKey]:
         """Create a checkpoint config in the store using a store_backend-specific key.
diff --git a/great_expectations/rule_based_profiler/rule_based_profiler.py b/great_expectations/rule_based_profiler/rule_based_profiler.py
index 8bb4b2302..7cc45255f 100644
--- a/great_expectations/rule_based_profiler/rule_based_profiler.py
+++ b/great_expectations/rule_based_profiler/rule_based_profiler.py
@@ -1057,7 +1057,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
         batch_list: Optional[List[Batch]] = None,
         batch_request: Optional[Union[BatchRequestBase, dict]] = None,
         name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        id: Optional[str] = None,
         variables: Optional[dict] = None,
         rules: Optional[dict] = None,
     ) -> RuleBasedProfilerResult:
@@ -1065,7 +1065,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
             data_context=data_context,
             profiler_store=profiler_store,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
         return profiler.run(
@@ -1087,13 +1087,13 @@ class BaseRuleBasedProfiler(ConfigPeer):
         batch_list: Optional[List[Batch]] = None,
         batch_request: Optional[Union[BatchRequestBase, dict]] = None,
         name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        id: Optional[str] = None,
     ) -> RuleBasedProfilerResult:
         profiler: RuleBasedProfiler = RuleBasedProfiler.get_profiler(
             data_context=data_context,
             profiler_store=profiler_store,
             name=name,
-            ge_cloud_id=ge_cloud_id,
+            id=id,
         )
 
         rule: Rule
@@ -1188,19 +1188,19 @@ class BaseRuleBasedProfiler(ConfigPeer):
         data_context: AbstractDataContext,
         profiler_store: ProfilerStore,
         name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        id: Optional[str] = None,
     ) -> RuleBasedProfiler:
-        key = RuleBasedProfiler._construct_profiler_key(name=name, id=ge_cloud_id)
+        key = RuleBasedProfiler._construct_profiler_key(name=name, id=id)
         try:
             profiler_config: RuleBasedProfilerConfig = profiler_store.get(key=key)
         except gx_exceptions.InvalidKeyError as exc_ik:
-            id: Union[GXCloudIdentifier, ConfigurationIdentifier] = (
+            config_id: Union[GXCloudIdentifier, ConfigurationIdentifier] = (
                 key.configuration_key
                 if isinstance(key, ConfigurationIdentifier)
                 else key
             )
             raise gx_exceptions.ProfilerNotFoundError(
-                message=f'Non-existent Profiler configuration named "{id}".\n\nDetails: {exc_ik}'
+                message=f'Non-existent Profiler configuration named "{config_id}".\n\nDetails: {exc_ik}'
             )
 
         config: dict = profiler_config.to_json_dict()
@@ -1226,20 +1226,20 @@ class BaseRuleBasedProfiler(ConfigPeer):
     def delete_profiler(
         profiler_store: ProfilerStore,
         name: Optional[str] = None,
-        ge_cloud_id: Optional[str] = None,
+        id: Optional[str] = None,
     ) -> None:
-        key = RuleBasedProfiler._construct_profiler_key(name=name, id=ge_cloud_id)
+        key = RuleBasedProfiler._construct_profiler_key(name=name, id=id)
 
         try:
             profiler_store.remove_key(key=key)
         except (gx_exceptions.InvalidKeyError, KeyError) as exc_ik:
-            id = (
+            config_id = (
                 key.configuration_key
                 if isinstance(key, ConfigurationIdentifier)
                 else key
             )
             raise gx_exceptions.ProfilerNotFoundError(
-                message=f'Non-existent Profiler configuration named "{id}".\n\nDetails: {exc_ik}'
+                message=f'Non-existent Profiler configuration named "{config_id}".\n\nDetails: {exc_ik}'
             )
 
     @staticmethod


Test patch:
diff --git a/tests/data_context/migrator/conftest.py b/tests/data_context/migrator/conftest.py
index 783af4cb1..cca33463f 100644
--- a/tests/data_context/migrator/conftest.py
+++ b/tests/data_context/migrator/conftest.py
@@ -29,7 +29,7 @@ class StubUsageStats:
 
 
 class StubCheckpointStore:
-    def get_checkpoint(self, name: str, ge_cloud_id: Optional[str]) -> CheckpointConfig:
+    def get_checkpoint(self, name: str, id: Optional[str]) -> CheckpointConfig:
         return CheckpointConfig(name=name, class_name="Checkpoint")
 
 
diff --git a/tests/data_context/store/test_checkpoint_store.py b/tests/data_context/store/test_checkpoint_store.py
index bebb94286..62c9c6896 100644
--- a/tests/data_context/store/test_checkpoint_store.py
+++ b/tests/data_context/store/test_checkpoint_store.py
@@ -415,7 +415,7 @@ def test_delete_checkpoint_with_cloud_id(
 ) -> None:
     store, mock_backend = checkpoint_store_with_mock_backend
 
-    store.delete_checkpoint(ge_cloud_id="abc123")
+    store.delete_checkpoint(id="abc123")
 
     mock_backend.remove_key.assert_called_once_with(
         GXCloudIdentifier(
@@ -450,7 +450,7 @@ def test_get_checkpoint(
     store, mock_backend = checkpoint_store_with_mock_backend
     mock_backend.get.return_value = checkpoint_config
 
-    checkpoint = store.get_checkpoint(name=checkpoint_config["name"], ge_cloud_id=None)
+    checkpoint = store.get_checkpoint(name=checkpoint_config["name"], id=None)
 
     actual_checkpoint_config = checkpoint.to_json_dict()
     for key, val in checkpoint_config.items():
@@ -468,7 +468,7 @@ def test_get_checkpoint_with_nonexistent_checkpoint_raises_error(
     mock_backend.get.side_effect = _raise_key_error
 
     with pytest.raises(gx_exceptions.CheckpointNotFoundError) as e:
-        store.get_checkpoint(name="my_fake_checkpoint", ge_cloud_id=None)
+        store.get_checkpoint(name="my_fake_checkpoint", id=None)
 
     assert 'Non-existent Checkpoint configuration named "my_fake_checkpoint".' in str(
         e.value
@@ -489,7 +489,7 @@ def test_get_checkpoint_with_invalid_checkpoint_config_raises_error(
         "great_expectations.data_context.store.CheckpointStore.deserialize",
         side_effect=_raise_validation_error,
     ), pytest.raises(gx_exceptions.InvalidCheckpointConfigError) as e:
-        store.get_checkpoint(name="my_fake_checkpoint", ge_cloud_id=None)
+        store.get_checkpoint(name="my_fake_checkpoint", id=None)
 
     assert "Invalid Checkpoint configuration" in str(e.value)
 
@@ -504,7 +504,7 @@ def test_get_checkpoint_with_invalid_legacy_checkpoint_raises_error(
     )  # Defaults to empty LegacyCheckpoint
 
     with pytest.raises(gx_exceptions.CheckpointError) as e:
-        store.get_checkpoint(name="my_checkpoint", ge_cloud_id=None)
+        store.get_checkpoint(name="my_checkpoint", id=None)
 
     assert (
         "Attempt to instantiate LegacyCheckpoint with insufficient and/or incorrect arguments"
@@ -522,7 +522,7 @@ def test_add_checkpoint(
     context._usage_statistics_handler = mock.MagicMock()
     checkpoint = Checkpoint(name="my_checkpoint", data_context=context)
 
-    store.add_checkpoint(checkpoint=checkpoint, name="my_checkpoint", ge_cloud_id=None)
+    store.add_checkpoint(checkpoint=checkpoint, name="my_checkpoint", id=None)
 
     mock_backend.set.assert_called_once_with(
         ("my_checkpoint",),
diff --git a/tests/data_context/test_data_context_deprecation.py b/tests/data_context/test_data_context_deprecation.py
index 4767881af..f20996cd5 100644
--- a/tests/data_context/test_data_context_deprecation.py
+++ b/tests/data_context/test_data_context_deprecation.py
@@ -4,12 +4,13 @@ from unittest import mock
 
 import pytest
 
+from great_expectations.data_context.data_context import (
+    AbstractDataContext,
+    CloudDataContext,
+)
 from great_expectations.data_context.data_context.base_data_context import (
     _resolve_cloud_args as base_data_context_resolver,
 )
-from great_expectations.data_context.data_context.cloud_data_context import (
-    CloudDataContext,
-)
 from great_expectations.data_context.data_context.data_context import (
     _resolve_cloud_args as data_context_resolver,
 )
@@ -245,6 +246,39 @@ def test_get_context_resolve_cloud_args(
     assert actual_resolved_args == expected_resolved_args
 
 
+@pytest.mark.unit
+@pytest.mark.cloud
+@pytest.mark.parametrize(
+    "id, ge_cloud_id, expected",
+    [
+        pytest.param("abc123", None, "abc123", id="only id"),
+        pytest.param(None, "def456", "def456", id="only ge_cloud_id"),
+        pytest.param(None, None, None, id="neither id nor ge_cloud_id"),
+    ],
+)
+def test_data_context__resolve_id_and_ge_cloud_id_success(
+    id: str | None, ge_cloud_id: str | None, expected: str | None
+):
+    resolved = AbstractDataContext._resolve_id_and_ge_cloud_id(
+        id=id, ge_cloud_id=ge_cloud_id
+    )
+    assert resolved == expected
+
+
+@pytest.mark.unit
+@pytest.mark.cloud
+def test_data_context__resolve_id_and_ge_cloud_id_failure():
+    id = "abc123"
+    ge_cloud_id = "def456"
+
+    with pytest.raises(ValueError) as e:
+        _ = AbstractDataContext._resolve_id_and_ge_cloud_id(
+            id=id, ge_cloud_id=ge_cloud_id
+        )
+
+    assert "either id or ge_cloud_id (not both)" in str(e.value)
+
+
 @pytest.mark.unit
 @pytest.mark.parametrize(
     "method_name, required_args",
diff --git a/tests/data_context/test_data_context_profilers.py b/tests/data_context/test_data_context_profilers.py
index da080c3ab..76ea63df2 100644
--- a/tests/data_context/test_data_context_profilers.py
+++ b/tests/data_context/test_data_context_profilers.py
@@ -122,7 +122,7 @@ def test_save_profiler(
             profiler=profiler_config_with_placeholder_args,
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
 
     with mock.patch(
@@ -133,6 +133,6 @@ def test_save_profiler(
             data_context=mock_data_context,
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
     assert isinstance(profiler, RuleBasedProfiler)
diff --git a/tests/rule_based_profiler/test_rule_based_profiler.py b/tests/rule_based_profiler/test_rule_based_profiler.py
index e6ce720f7..de458b6d1 100644
--- a/tests/rule_based_profiler/test_rule_based_profiler.py
+++ b/tests/rule_based_profiler/test_rule_based_profiler.py
@@ -1095,7 +1095,7 @@ def test_get_profiler_with_too_many_args_raises_error(
             data_context=mock_data_context,
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id="my_ge_cloud_id",
+            id="my_ge_cloud_id",
         )
 
     assert "either name or id" in str(e.value)
@@ -1268,7 +1268,7 @@ def test_get_profiler(
             data_context=mock_data_context,
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
 
     assert isinstance(profiler, RuleBasedProfiler)
@@ -1284,7 +1284,7 @@ def test_get_profiler_non_existent_profiler_raises_error(
             data_context=mock_data_context,
             profiler_store=empty_profiler_store,
             name="my_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
 
     assert "Non-existent Profiler" in str(e.value)
@@ -1300,7 +1300,7 @@ def test_delete_profiler(
         RuleBasedProfiler.delete_profiler(
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
 
     assert mock_remove_key.call_count == 1
@@ -1317,7 +1317,7 @@ def test_delete_profiler_with_too_many_args_raises_error(
         RuleBasedProfiler.delete_profiler(
             profiler_store=populated_profiler_store,
             name="my_profiler",
-            ge_cloud_id="my_ge_cloud_id",
+            id="my_ge_cloud_id",
         )
 
     assert "either name or id" in str(e.value)
@@ -1331,7 +1331,7 @@ def test_delete_profiler_non_existent_profiler_raises_error(
         RuleBasedProfiler.delete_profiler(
             profiler_store=populated_profiler_store,
             name="my_non_existent_profiler",
-            ge_cloud_id=None,
+            id=None,
         )
 
     assert "Non-existent Profiler" in str(e.value)