Repository: great-expectations/great_expectations
Pull request #7081 (2023-02-08 16:06:50)

[MAINTENANCE] Refactor `Store` and `StoreBackend` to leverage new CRUD methods
Changes proposed in this pull request:
- Add `add`, `update`, and `add_or_update` logic to `Store` and `StoreBackend`
- Refactor context CRUD to leverage these changes


### 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 made corresponding changes to the documentation
- [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.


netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | c4a52c94448ae1db4fb2c42b4da6c515a12d9b1b |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63e3c8aa3e2e9e000858ea12 |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=7081&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=7081&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/7081/488418f0/c4a52c94448ae1db4fb2c42b4da6c515a12d9b1b.svg)](https://app.codesee.io/r/reviews?pr=7081&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 9684af8d2..308f9e020 100644
--- a/great_expectations/data_context/data_context/abstract_data_context.py
+++ b/great_expectations/data_context/data_context/abstract_data_context.py
@@ -56,9 +56,6 @@ from great_expectations.core.config_provider import (
     _EnvironmentConfigurationProvider,
     _RuntimeEnvironmentConfigurationProvider,
 )
-from great_expectations.core.expectation_configuration import (
-    ExpectationConfiguration,  # noqa: TCH001
-)
 from great_expectations.core.expectation_validation_result import get_metric_kwargs_id
 from great_expectations.core.id_dict import BatchKwargs
 from great_expectations.core.run_identifier import RunIdentifier
@@ -73,19 +70,7 @@ from great_expectations.data_asset import DataAsset
 from great_expectations.data_context.config_validator.yaml_config_validator import (
     _YamlConfigValidator,
 )
-from great_expectations.data_context.data_context_variables import (
-    DataContextVariables,  # noqa: TCH001
-)
 from great_expectations.data_context.store import Store, TupleStoreBackend
-from great_expectations.data_context.store.expectations_store import (
-    ExpectationsStore,  # noqa: TCH001
-)
-from great_expectations.data_context.store.profiler_store import (
-    ProfilerStore,  # noqa: TCH001
-)
-from great_expectations.data_context.store.validations_store import (
-    ValidationsStore,  # noqa: TCH001
-)
 from great_expectations.data_context.templates import CONFIG_VARIABLES_TEMPLATE
 from great_expectations.data_context.types.base import (
     CURRENT_GX_CONFIG_VERSION,
@@ -123,7 +108,6 @@ from great_expectations.datasource.datasource_serializer import (
     NamedDatasourceSerializer,
 )
 from great_expectations.datasource.new_datasource import BaseDatasource, Datasource
-from great_expectations.execution_engine import ExecutionEngine  # noqa: TCH001
 from great_expectations.experimental.datasources.config import GxConfig
 from great_expectations.experimental.datasources.interfaces import Batch as XBatch
 from great_expectations.experimental.datasources.interfaces import (
@@ -131,10 +115,6 @@ from great_expectations.experimental.datasources.interfaces import (
 )
 from great_expectations.experimental.datasources.sources import _SourceFactories
 from great_expectations.profile.basic_dataset_profiler import BasicDatasetProfiler
-from great_expectations.rule_based_profiler.config.base import (
-    RuleBasedProfilerConfig,
-    ruleBasedProfilerConfigSchema,
-)
 from great_expectations.rule_based_profiler.data_assistant.data_assistant_dispatcher import (
     DataAssistantDispatcher,
 )
@@ -162,13 +142,25 @@ except ImportError:
 if TYPE_CHECKING:
     from great_expectations.checkpoint import Checkpoint
     from great_expectations.checkpoint.types.checkpoint_result import CheckpointResult
+    from great_expectations.core.expectation_configuration import (
+        ExpectationConfiguration,
+    )
+    from great_expectations.data_context.data_context_variables import (
+        DataContextVariables,
+    )
     from great_expectations.data_context.store import (
         CheckpointStore,
         EvaluationParameterStore,
     )
+    from great_expectations.data_context.store.expectations_store import (
+        ExpectationsStore,
+    )
+    from great_expectations.data_context.store.profiler_store import ProfilerStore
+    from great_expectations.data_context.store.validations_store import ValidationsStore
     from great_expectations.data_context.types.resource_identifiers import (
         GXCloudIdentifier,
     )
+    from great_expectations.execution_engine import ExecutionEngine
     from great_expectations.render.renderer.site_builder import SiteBuilder
     from great_expectations.rule_based_profiler import RuleBasedProfilerResult
     from great_expectations.validation_operators.validation_operators import (
@@ -2742,16 +2734,14 @@ class AbstractDataContext(ConfigPeer, ABC):
 
         expectation_suite_name = expectation_suite.expectation_suite_name
         key = ExpectationSuiteIdentifier(expectation_suite_name=expectation_suite_name)
-        if (
-            self.expectations_store.has_key(key)  # noqa: W601
-            and not overwrite_existing
-        ):
-            raise gx_exceptions.DataContextError(
-                f"expectation_suite with name {expectation_suite_name} already exists."
-                " If you would like to overwrite this expectation_suite, please delete or"
-                " update it using `delete_expectation_suite` or `update_expectation_suite`, respectively."
-            )
-        self.expectations_store.set(key, expectation_suite, **kwargs)
+
+        persistence_fn: Callable
+        if overwrite_existing:
+            persistence_fn = self.expectations_store.add_or_update
+        else:
+            persistence_fn = self.expectations_store.add
+
+        persistence_fn(key=key, value=expectation_suite, **kwargs)
         return expectation_suite
 
     @public_api
@@ -2770,15 +2760,8 @@ class AbstractDataContext(ConfigPeer, ABC):
         """
         expectation_suite_name: str = expectation_suite.expectation_suite_name
         key = ExpectationSuiteIdentifier(expectation_suite_name=expectation_suite_name)
-        if not self.expectations_store.has_key(key):  # noqa: W601
-            raise gx_exceptions.DataContextError(
-                f"expectation_suite with name {expectation_suite_name} does not exist."
-            )
-
-        return self._add_expectation_suite(
-            expectation_suite=expectation_suite,
-            overwrite_existing=True,
-        )
+        self.expectations_store.update(key=key, value=expectation_suite)
+        return expectation_suite
 
     @overload
     def add_or_update_expectation_suite(
@@ -3001,58 +2984,16 @@ class AbstractDataContext(ConfigPeer, ABC):
         Returns:
             The persisted Profiler constructed by the input arguments.
         """
-        return self._add_profiler(
+        return RuleBasedProfiler.add_profiler(
+            data_context=self,
+            profiler_store=self.profiler_store,
             name=name,
-            id=None,
             config_version=config_version,
             rules=rules,
             variables=variables,
             profiler=profiler,
         )
 
-    def _add_profiler(
-        self,
-        name: str | None,
-        id: str | None,
-        config_version: float | None,
-        rules: dict[str, dict] | None,
-        variables: dict | None,
-        profiler: RuleBasedProfiler | None,
-    ) -> RuleBasedProfiler:
-        if not (
-            (profiler is None)
-            ^ all(arg is None for arg in (name, config_version, rules))
-        ):
-            raise ValueError(
-                "Must either pass in an existing profiler or individual constructor arguments (but not both)"
-            )
-
-        if profiler:
-            config = profiler.config
-        else:
-            config_data = {
-                "name": name,
-                "id": id,
-                "config_version": config_version,
-                "rules": rules,
-                "variables": variables,
-            }
-
-            # Roundtrip through schema validation to remove any illegal fields add/or restore any missing fields.
-            validated_config: dict = ruleBasedProfilerConfigSchema.load(config_data)
-            profiler_config: dict = ruleBasedProfilerConfigSchema.dump(validated_config)
-            profiler_config.pop("class_name")
-            profiler_config.pop("module_name")
-
-            config = RuleBasedProfilerConfig(**profiler_config)
-
-        profiler = RuleBasedProfiler.add_profiler(
-            config=config,
-            data_context=self,
-            profiler_store=self.profiler_store,
-        )
-        return profiler
-
     @public_api
     @new_argument(
         argument_name="id",
@@ -3133,9 +3074,9 @@ class AbstractDataContext(ConfigPeer, ABC):
             ProfilerNotFoundError: A profiler with the given name/id does not already exist.
         """
         return RuleBasedProfiler.update_profiler(
-            profiler=profiler,
             profiler_store=self.profiler_store,
             data_context=self,
+            profiler=profiler,
         )
 
     @overload
@@ -3192,7 +3133,9 @@ class AbstractDataContext(ConfigPeer, ABC):
         Returns:
             A new Profiler or an updated one (depending on whether or not it existed before this method call).
         """
-        return self._add_profiler(
+        return RuleBasedProfiler.add_or_update_profiler(
+            data_context=self,
+            profiler_store=self.profiler_store,
             name=name,
             id=id,
             config_version=config_version,
diff --git a/great_expectations/data_context/store/_store_backend.py b/great_expectations/data_context/store/_store_backend.py
index 2a46b1d8c..c973a6068 100644
--- a/great_expectations/data_context/store/_store_backend.py
+++ b/great_expectations/data_context/store/_store_backend.py
@@ -130,6 +130,32 @@ class StoreBackend(metaclass=ABCMeta):
             logger.debug(str(e))
             raise StoreBackendError("ValueError while calling _set on store backend.")
 
+    def add(self, key, value, **kwargs):
+        """
+        Essentially `set` but validates that a given key-value pair does not already exist.
+        """
+        if self.has_key(key):
+            raise StoreBackendError(f"Store already has the following key: {key}.")
+        return self.set(key=key, value=value, **kwargs)
+
+    def update(self, key, value, **kwargs):
+        """
+        Essentially `set` but validates that a given key-value pair does already exist.
+        """
+        if not self.has_key(key):
+            raise StoreBackendError(
+                f"Store does not have a value associated the following key: {key}."
+            )
+        return self.set(key=key, value=value, **kwargs)
+
+    def add_or_update(self, key, value, **kwargs):
+        """
+        Conditionally calls `add` or `update` based on the presence of the given key.
+        """
+        if self.has_key(key):
+            return self.update(key=key, value=value, **kwargs)
+        return self.add(key=key, value=value, **kwargs)
+
     def move(self, source_key, dest_key, **kwargs):
         self._validate_key(source_key)
         self._validate_key(dest_key)
diff --git a/great_expectations/data_context/store/checkpoint_store.py b/great_expectations/data_context/store/checkpoint_store.py
index a67497f90..4554e4453 100644
--- a/great_expectations/data_context/store/checkpoint_store.py
+++ b/great_expectations/data_context/store/checkpoint_store.py
@@ -5,7 +5,7 @@ import logging
 import os
 import random
 import uuid
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
 
 from marshmallow import ValidationError
 
@@ -21,9 +21,9 @@ from great_expectations.data_context.types.refs import (
     GXCloudIDAwareRef,
     GXCloudResourceRef,
 )
-from great_expectations.data_context.types.resource_identifiers import (
-    ConfigurationIdentifier,  # noqa: TCH001
-    GXCloudIdentifier,  # noqa: TCH001
+from great_expectations.data_context.types.resource_identifiers import (  # noqa: TCH001
+    ConfigurationIdentifier,
+    GXCloudIdentifier,
 )
 
 if TYPE_CHECKING:
@@ -177,42 +177,82 @@ class CheckpointStore(ConfigurationStore):
         return checkpoint_config
 
     def add_checkpoint(self, checkpoint: Checkpoint) -> Checkpoint:
+        """Persist a stand-alone Checkpoint object.
+
+        Args:
+            checkpoint: The Checkpoint to persist.
+
+        Returns:
+            The persisted Checkpoint (possibly modified state based on store backend).
+
+        Raises:
+            CheckpointError: If a Checkpoint with the given name already exists.
+        """
         key = self._construct_key_from_checkpoint(checkpoint)
-        return self._add_checkpoint(key=key, checkpoint=checkpoint)
+        try:
+            return self._persist_checkpoint(
+                key=key, checkpoint=checkpoint, persistence_fn=self.add
+            )
+        except gx_exceptions.StoreBackendError:
+            raise gx_exceptions.CheckpointError(
+                f"A Checkpoint named {checkpoint.name} already exists."
+            )
 
     def update_checkpoint(self, checkpoint: Checkpoint) -> Checkpoint:
+        """Use a stand-alone Checkpoint object to update a persisted value.
+
+        Args:
+            checkpoint: The Checkpoint to use for updating.
+
+        Returns:
+            The persisted Checkpoint (possibly modified state based on store backend).
+
+        Raises:
+            CheckpointNotFoundError: If a Checkpoint with the given name does not exist in the store.
+        """
         key = self._construct_key_from_checkpoint(checkpoint)
-        if not self.has_key(key):
+        try:
+            return self._persist_checkpoint(
+                key=key, checkpoint=checkpoint, persistence_fn=self.update
+            )
+        except gx_exceptions.StoreBackendError:
             raise gx_exceptions.CheckpointNotFoundError(
-                f"Could not find a Checkpoint named {checkpoint.name}"
+                f"Could not find an existing Checkpoint named {checkpoint.name}."
             )
-        return self._add_checkpoint(key=key, checkpoint=checkpoint)
 
     def add_or_update_checkpoint(self, checkpoint: Checkpoint) -> Checkpoint:
+        """Use a stand-alone Checkpoint object to either add or update a persisted value.
+
+        Args:
+            checkpoint: The Checkpoint to use for adding or updating.
+
+        Returns:
+            The persisted Checkpoint (possibly modified state based on store backend).
+        """
         key = self._construct_key_from_checkpoint(checkpoint)
-        if self.has_key(key):
-            return self.update_checkpoint(checkpoint)
-        return self._add_checkpoint(key=key, checkpoint=checkpoint)
+        return self._persist_checkpoint(
+            key=key, checkpoint=checkpoint, persistence_fn=self.add_or_update
+        )
 
     def _construct_key_from_checkpoint(
         self, checkpoint: Checkpoint
-    ) -> Union[GXCloudIdentifier, ConfigurationIdentifier]:
+    ) -> GXCloudIdentifier | ConfigurationIdentifier:
         name = checkpoint.name
         id = checkpoint.ge_cloud_id
         if id:
             return self.determine_key(ge_cloud_id=str(id))
         return self.determine_key(name=name)
 
-    def _add_checkpoint(
+    def _persist_checkpoint(
         self,
         key: GXCloudIdentifier | ConfigurationIdentifier,
         checkpoint: Checkpoint,
+        persistence_fn: Callable,
     ) -> Checkpoint:
-        checkpoint_config = checkpoint.get_config()
-        checkpoint_ref = self.set(key=key, value=checkpoint_config)  # type: ignore[func-returns-value]
+        checkpoint_ref = persistence_fn(key=key, value=checkpoint.get_config())
         if isinstance(checkpoint_ref, GXCloudIDAwareRef):
             cloud_id = checkpoint_ref.cloud_id
-            checkpoint.ge_cloud_id = uuid.UUID(cloud_id)  # type: ignore[misc]
+            checkpoint.config.ge_cloud_id = uuid.UUID(cloud_id)
         return checkpoint
 
     def create(self, checkpoint_config: CheckpointConfig) -> Optional[DataContextKey]:
diff --git a/great_expectations/data_context/store/expectations_store.py b/great_expectations/data_context/store/expectations_store.py
index 537378345..69f9e8ac3 100644
--- a/great_expectations/data_context/store/expectations_store.py
+++ b/great_expectations/data_context/store/expectations_store.py
@@ -2,6 +2,7 @@ import random
 import uuid
 from typing import Dict
 
+import great_expectations.exceptions as gx_exceptions
 from great_expectations.core import ExpectationSuite
 from great_expectations.core.expectation_suite import ExpectationSuiteSchema
 from great_expectations.data_context.cloud_constants import GXCloudRESTResource
@@ -177,6 +178,22 @@ class ExpectationsStore(Store):
 
         return suite_dict
 
+    def add(self, key, value, **kwargs):
+        try:
+            super().add(key=key, value=value, **kwargs)
+        except gx_exceptions.StoreBackendError:
+            raise gx_exceptions.ExpectationSuiteError(
+                f"An ExpectationSuite named {value.expectation_suite_name} already exists."
+            )
+
+    def update(self, key, value, **kwargs):
+        try:
+            super().update(key=key, value=value, **kwargs)
+        except gx_exceptions.StoreBackendError:
+            raise gx_exceptions.ExpectationSuiteError(
+                f"Could not find an existing ExpectationSuite named {value.expectation_suite_name}."
+            )
+
     def get(self, key) -> ExpectationSuite:
         return super().get(key)  # type: ignore[return-value]
 
diff --git a/great_expectations/data_context/store/profiler_store.py b/great_expectations/data_context/store/profiler_store.py
index d71e0665a..5da7e2a1b 100644
--- a/great_expectations/data_context/store/profiler_store.py
+++ b/great_expectations/data_context/store/profiler_store.py
@@ -2,6 +2,7 @@ import random
 import uuid
 from typing import Union
 
+import great_expectations.exceptions as gx_exceptions
 from great_expectations.data_context.cloud_constants import GXCloudRESTResource
 from great_expectations.data_context.store.configuration_store import ConfigurationStore
 from great_expectations.data_context.types.resource_identifiers import (
@@ -72,3 +73,19 @@ class ProfilerStore(ConfigurationStore):
         profiler_config_dict["id"] = ge_cloud_profiler_id
 
         return profiler_config_dict
+
+    def add(self, key, value, **kwargs):
+        try:
+            super().add(key=key, value=value, **kwargs)
+        except gx_exceptions.StoreBackendError:
+            raise gx_exceptions.ProfilerError(
+                f"A Profiler named {value.name} already exists."
+            )
+
+    def update(self, key, value, **kwargs):
+        try:
+            super().update(key=key, value=value, **kwargs)
+        except gx_exceptions.StoreBackendError:
+            raise gx_exceptions.ProfilerNotFoundError(
+                f"Could not find an existing Profiler named {value.name}."
+            )
diff --git a/great_expectations/data_context/store/store.py b/great_expectations/data_context/store/store.py
index f5c5f9d85..7e87fc822 100644
--- a/great_expectations/data_context/store/store.py
+++ b/great_expectations/data_context/store/store.py
@@ -183,6 +183,33 @@ class Store:
             self.key_to_tuple(key), self.serialize(value), **kwargs
         )
 
+    def add(self, key: DataContextKey, value: Any, **kwargs) -> None:
+        """
+        Essentially `set` but validates that a given key-value pair does not already exist.
+        """
+        self._validate_key(key)
+        return self._store_backend.add(
+            self.key_to_tuple(key), self.serialize(value), **kwargs
+        )
+
+    def update(self, key: DataContextKey, value: Any, **kwargs) -> None:
+        """
+        Essentially `set` but validates that a given key-value pair does already exist.
+        """
+        self._validate_key(key)
+        return self._store_backend.update(
+            self.key_to_tuple(key), self.serialize(value), **kwargs
+        )
+
+    def add_or_update(self, key: DataContextKey, value: Any, **kwargs) -> None:
+        """
+        Conditionally calls `add` or `update` based on the presence of the given key.
+        """
+        self._validate_key(key)
+        return self._store_backend.add_or_update(
+            self.key_to_tuple(key), self.serialize(value), **kwargs
+        )
+
     def list_keys(self) -> List[DataContextKey]:
         keys_without_store_backend_id = [
             key
diff --git a/great_expectations/exceptions/exceptions.py b/great_expectations/exceptions/exceptions.py
index a7bd23ea3..7076d619e 100644
--- a/great_expectations/exceptions/exceptions.py
+++ b/great_expectations/exceptions/exceptions.py
@@ -36,6 +36,10 @@ class DataContextError(GreatExpectationsError):
     pass
 
 
+class ExpectationSuiteError(DataContextError):
+    pass
+
+
 class CheckpointError(DataContextError):
     pass
 
diff --git a/great_expectations/rule_based_profiler/rule_based_profiler.py b/great_expectations/rule_based_profiler/rule_based_profiler.py
index 10a1c0084..573394ad5 100644
--- a/great_expectations/rule_based_profiler/rule_based_profiler.py
+++ b/great_expectations/rule_based_profiler/rule_based_profiler.py
@@ -15,10 +15,6 @@ from great_expectations.core.batch import (
     batch_request_contains_batch_data,
 )
 from great_expectations.core.config_peer import ConfigPeer
-from great_expectations.core.domain import Domain  # noqa: TCH001
-from great_expectations.core.expectation_configuration import (
-    ExpectationConfiguration,  # noqa: TCH001
-)
 from great_expectations.core.usage_statistics.events import UsageStatsEvents
 from great_expectations.core.usage_statistics.usage_statistics import (
     UsageStatisticsHandler,
@@ -46,9 +42,7 @@ from great_expectations.rule_based_profiler.config.base import (
     domainBuilderConfigSchema,
     expectationConfigurationBuilderConfigSchema,
     parameterBuilderConfigSchema,
-)
-from great_expectations.rule_based_profiler.domain_builder.domain_builder import (
-    DomainBuilder,  # noqa: TCH001
+    ruleBasedProfilerConfigSchema,
 )
 from great_expectations.rule_based_profiler.expectation_configuration_builder import (
     ExpectationConfigurationBuilder,
@@ -60,10 +54,6 @@ from great_expectations.rule_based_profiler.helpers.configuration_reconciliation
     ReconciliationStrategy,
     reconcile_rule_variables,
 )
-from great_expectations.rule_based_profiler.helpers.runtime_environment import (
-    RuntimeEnvironmentDomainTypeDirectives,  # noqa: TCH001
-    RuntimeEnvironmentVariablesDirectives,  # noqa: TCH001
-)
 from great_expectations.rule_based_profiler.helpers.util import (
     convert_variables_to_dict,
 )
@@ -81,8 +71,19 @@ from great_expectations.rule_based_profiler.rule.rule_state import RuleState
 from great_expectations.util import filter_properties_dict
 
 if TYPE_CHECKING:
+    from great_expectations.core.domain import Domain
+    from great_expectations.core.expectation_configuration import (
+        ExpectationConfiguration,
+    )
     from great_expectations.data_context import AbstractDataContext
     from great_expectations.data_context.store.profiler_store import ProfilerStore
+    from great_expectations.rule_based_profiler.domain_builder.domain_builder import (
+        DomainBuilder,
+    )
+    from great_expectations.rule_based_profiler.helpers.runtime_environment import (
+        RuntimeEnvironmentDomainTypeDirectives,
+        RuntimeEnvironmentVariablesDirectives,
+    )
 
 
 logger = logging.getLogger(__name__)
@@ -583,8 +584,9 @@ class BaseRuleBasedProfiler(ConfigPeer):
         effective_rules.update(override_rules)
         return effective_rules
 
-    @staticmethod
+    @classmethod
     def _reconcile_rule_config(
+        cls,
         existing_rules: Dict[str, Rule],
         rule_name: str,
         rule_config: dict,
@@ -631,7 +633,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
 
             domain_builder_config: dict = rule_config.get("domain_builder", {})
             effective_domain_builder_config: dict = (
-                RuleBasedProfiler._reconcile_rule_domain_builder_config(
+                cls._reconcile_rule_domain_builder_config(
                     domain_builder=rule.domain_builder,
                     domain_builder_config=domain_builder_config,
                     reconciliation_strategy=reconciliation_directives.domain_builder,
@@ -643,7 +645,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
             )
             effective_parameter_builder_configs: Optional[
                 List[dict]
-            ] = RuleBasedProfiler._reconcile_rule_parameter_builder_configs(
+            ] = cls._reconcile_rule_parameter_builder_configs(
                 rule=rule,
                 parameter_builder_configs=parameter_builder_configs,
                 reconciliation_strategy=reconciliation_directives.parameter_builder,
@@ -654,7 +656,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
             )
             effective_expectation_configuration_builder_configs: List[
                 dict
-            ] = RuleBasedProfiler._reconcile_rule_expectation_configuration_builder_configs(
+            ] = cls._reconcile_rule_expectation_configuration_builder_configs(
                 rule=rule,
                 expectation_configuration_builder_configs=expectation_configuration_builder_configs,
                 reconciliation_strategy=reconciliation_directives.expectation_configuration_builder,
@@ -1052,8 +1054,9 @@ class BaseRuleBasedProfiler(ConfigPeer):
 
         return dest_property_value
 
-    @staticmethod
+    @classmethod
     def run_profiler(
+        cls,
         data_context: AbstractDataContext,
         profiler_store: ProfilerStore,
         batch_list: Optional[List[Batch]] = None,
@@ -1063,7 +1066,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
         variables: Optional[dict] = None,
         rules: Optional[dict] = None,
     ) -> RuleBasedProfilerResult:
-        profiler: RuleBasedProfiler = RuleBasedProfiler.get_profiler(
+        profiler: RuleBasedProfiler = cls.get_profiler(
             data_context=data_context,
             profiler_store=profiler_store,
             name=name,
@@ -1082,8 +1085,9 @@ class BaseRuleBasedProfiler(ConfigPeer):
             comment=None,
         )
 
-    @staticmethod
+    @classmethod
     def run_profiler_on_data(
+        cls,
         data_context: AbstractDataContext,
         profiler_store: ProfilerStore,
         batch_list: Optional[List[Batch]] = None,
@@ -1091,7 +1095,7 @@ class BaseRuleBasedProfiler(ConfigPeer):
         name: Optional[str] = None,
         id: Optional[str] = None,
     ) -> RuleBasedProfilerResult:
-        profiler: RuleBasedProfiler = RuleBasedProfiler.get_profiler(
+        profiler: RuleBasedProfiler = cls.get_profiler(
             data_context=data_context,
             profiler_store=profiler_store,
             name=name,
@@ -1115,17 +1119,101 @@ class BaseRuleBasedProfiler(ConfigPeer):
             comment=None,
         )
 
-    @staticmethod
+    @classmethod
     def add_profiler(
-        config: RuleBasedProfilerConfig,
+        cls,
         data_context: AbstractDataContext,
         profiler_store: ProfilerStore,
+        name: str | None = None,
+        id: str | None = None,
+        config_version: float | None = None,
+        rules: dict[str, dict] | None = None,
+        variables: dict | None = None,
+        profiler: RuleBasedProfiler | None = None,
+    ) -> RuleBasedProfiler:
+        return cls._persist_profiler(
+            data_context=data_context,
+            persistence_fn=profiler_store.add,
+            name=name,
+            id=id,
+            config_version=config_version,
+            rules=rules,
+            variables=variables,
+            profiler=profiler,
+        )
+
+    @classmethod
+    def update_profiler(
+        cls,
+        profiler_store: ProfilerStore,
+        data_context: AbstractDataContext,
+        name: str | None = None,
+        id: str | None = None,
+        config_version: float | None = None,
+        rules: dict[str, dict] | None = None,
+        variables: dict | None = None,
+        profiler: RuleBasedProfiler | None = None,
+    ) -> RuleBasedProfiler:
+        return cls._persist_profiler(
+            data_context=data_context,
+            persistence_fn=profiler_store.update,
+            name=name,
+            id=id,
+            config_version=config_version,
+            rules=rules,
+            variables=variables,
+            profiler=profiler,
+        )
+
+    @classmethod
+    def add_or_update_profiler(
+        cls,
+        data_context: AbstractDataContext,
+        profiler_store: ProfilerStore,
+        name: str | None = None,
+        id: str | None = None,
+        config_version: float | None = None,
+        rules: dict[str, dict] | None = None,
+        variables: dict | None = None,
+        profiler: RuleBasedProfiler | None = None,
+    ) -> RuleBasedProfiler:
+        return cls._persist_profiler(
+            data_context=data_context,
+            persistence_fn=profiler_store.add_or_update,
+            name=name,
+            id=id,
+            config_version=config_version,
+            rules=rules,
+            variables=variables,
+            profiler=profiler,
+        )
+
+    @classmethod
+    def _persist_profiler(
+        cls,
+        data_context: AbstractDataContext,
+        persistence_fn: Callable,
+        name: str | None = None,
+        id: str | None = None,
+        config_version: float | None = None,
+        rules: dict[str, dict] | None = None,
+        variables: dict | None = None,
+        profiler: RuleBasedProfiler | None = None,
     ) -> RuleBasedProfiler:
+        config = RuleBasedProfiler._resolve_profiler_config_for_store(
+            name=name,
+            id=id,
+            config_version=config_version,
+            rules=rules,
+            variables=variables,
+            profiler=profiler,
+        )
+
         if not RuleBasedProfiler._check_validity_of_batch_requests_in_config(
             config=config
         ):
             raise gx_exceptions.InvalidConfigError(
-                f'batch_data found in batch_request cannot be saved to ProfilerStore "{profiler_store.store_name}"'
+                "batch_data found in batch_request cannot be saved to ProfilerStore"
             )
 
         # Chetan - 20220204 - AbstractDataContext to be removed once it can be decoupled from RBP
@@ -1153,12 +1241,54 @@ class BaseRuleBasedProfiler(ConfigPeer):
                 configuration_key=config.name,
             )
 
-        response = profiler_store.set(key=key, value=config)
+        response = persistence_fn(key=key, value=config)
         if isinstance(response, GXCloudResourceRef):
             new_profiler.ge_cloud_id = response.cloud_id
 
         return new_profiler
 
+    @staticmethod
+    def _resolve_profiler_config_for_store(
+        name: str | None = None,
+        id: str | None = None,
+        config_version: float | None = None,
+        rules: dict[str, dict] | None = None,
+        variables: dict | None = None,
+        profiler: RuleBasedProfiler | None = None,
+    ) -> RuleBasedProfilerConfig:
+        if not (
+            (profiler is None)
+            ^ all(arg is None for arg in (name, config_version, rules))
+        ):
+            raise ValueError(
+                "Must either pass in an existing profiler or individual constructor arguments (but not both)"
+            )
+
+        if profiler:
+            config = profiler.config
+            # Recognize any changes made to the runtime object
+            config.config_version = profiler.config_version
+            config.rules = {rule.name: rule.to_dict() for rule in profiler.rules}
+            config.variables = convert_variables_to_dict(profiler.variables)
+        else:
+            config_data = {
+                "name": name,
+                "id": id,
+                "config_version": config_version,
+                "rules": rules,
+                "variables": variables,
+            }
+
+            # Roundtrip through schema validation to remove any illegal fields add/or restore any missing fields.
+            validated_config: dict = ruleBasedProfilerConfigSchema.load(config_data)
+            profiler_config: dict = ruleBasedProfilerConfigSchema.dump(validated_config)
+            profiler_config.pop("class_name")
+            profiler_config.pop("module_name")
+
+            config = RuleBasedProfilerConfig(**profiler_config)
+
+        return config
+
     @staticmethod
     def _check_validity_of_batch_requests_in_config(
         config: RuleBasedProfilerConfig,
@@ -1185,14 +1315,15 @@ class BaseRuleBasedProfiler(ConfigPeer):
 
         return True
 
-    @staticmethod
+    @classmethod
     def get_profiler(
+        cls,
         data_context: AbstractDataContext,
         profiler_store: ProfilerStore,
         name: Optional[str] = None,
         id: Optional[str] = None,
     ) -> RuleBasedProfiler:
-        key = RuleBasedProfiler._construct_profiler_key(name=name, id=id)
+        key = cls._construct_profiler_key(name=name, id=id)
         try:
             profiler_config: RuleBasedProfilerConfig = profiler_store.get(key=key)
         except gx_exceptions.InvalidKeyError as exc_ik:
@@ -1224,13 +1355,14 @@ class BaseRuleBasedProfiler(ConfigPeer):
 
         return profiler
 
-    @staticmethod
+    @classmethod
     def delete_profiler(
+        cls,
         profiler_store: ProfilerStore,
         name: Optional[str] = None,
         id: Optional[str] = None,
     ) -> None:
-        key = RuleBasedProfiler._construct_profiler_key(name=name, id=id)
+        key = cls._construct_profiler_key(name=name, id=id)
 
         try:
             profiler_store.remove_key(key=key)
@@ -1244,43 +1376,6 @@ class BaseRuleBasedProfiler(ConfigPeer):
                 message=f'Non-existent Profiler configuration named "{config_id}".\n\nDetails: {exc_ik}'
             )
 
-    @staticmethod
-    def update_profiler(
-        profiler: RuleBasedProfiler,
-        profiler_store: ProfilerStore,
-        data_context: AbstractDataContext,
-    ) -> RuleBasedProfiler:
-        name = profiler.name
-        id = profiler.ge_cloud_id
-
-        key = RuleBasedProfiler._construct_profiler_key(name=name, id=id)
-        if not profiler_store.has_key(key):
-            msg = "Non-existent Profiler configuration "
-            if name:
-                msg += f"named {name}"
-            elif id:
-                msg += f"with id {id}"
-            raise gx_exceptions.ProfilerNotFoundError(message=msg)
-
-        # Prioritize deletion by id if present (providing both will result in a downstream assertion error)
-        if id:
-            RuleBasedProfiler.delete_profiler(
-                ge_cloud_id=id, profiler_store=profiler_store
-            )
-        else:
-            RuleBasedProfiler.delete_profiler(name=name, profiler_store=profiler_store)
-
-        config = profiler.config
-        config.config_version = profiler.config_version
-        config.rules = {rule.name: rule.to_dict() for rule in profiler.rules}
-        config.variables = convert_variables_to_dict(profiler.variables)
-
-        return RuleBasedProfiler.add_profiler(
-            config=config,
-            data_context=data_context,
-            profiler_store=profiler_store,
-        )
-
     @staticmethod
     def list_profilers(
         profiler_store: ProfilerStore,


Test patch:
diff --git a/tests/data_context/store/test_checkpoint_store.py b/tests/data_context/store/test_checkpoint_store.py
index f24f4f3a9..562989db2 100644
--- a/tests/data_context/store/test_checkpoint_store.py
+++ b/tests/data_context/store/test_checkpoint_store.py
@@ -525,7 +525,7 @@ def test_add_checkpoint(
 
     store.add_checkpoint(checkpoint=checkpoint)
 
-    mock_backend.set.assert_called_once_with(
+    mock_backend.add.assert_called_once_with(
         (checkpoint_name,),
         "name: my_checkpoint\nconfig_version:\nmodule_name: great_expectations.checkpoint\nclass_name: LegacyCheckpoint\n",
     )
diff --git a/tests/data_context/store/test_store.py b/tests/data_context/store/test_store.py
index d00732ca2..5a318aee7 100644
--- a/tests/data_context/store/test_store.py
+++ b/tests/data_context/store/test_store.py
@@ -2,7 +2,9 @@ import pytest
 
 import great_expectations.exceptions as gx_exceptions
 from great_expectations.core.configuration import AbstractConfig
+from great_expectations.core.data_context_key import StringKey
 from great_expectations.data_context.store.store import Store
+from great_expectations.exceptions.exceptions import StoreBackendError
 
 
 @pytest.mark.unit
@@ -69,3 +71,65 @@ def test_build_store_from_config_failure(store_config: dict, module_name: str):
             store_config=store_config,
             module_name=module_name,
         )
+
+
+@pytest.mark.unit
+def test_store_add_success():
+    store = Store()
+    key = StringKey("foo")
+    value = "bar"
+
+    store.add(key=key, value=value)
+    assert store.has_key(key)
+
+
+@pytest.mark.unit
+def test_store_add_failure():
+    store = Store()
+    key = StringKey("foo")
+    value = "bar"
+
+    store.add(key=key, value=value)
+    with pytest.raises(StoreBackendError) as e:
+        store.add(key=key, value=value)
+
+    assert "Store already has the following key" in str(e.value)
+
+
+@pytest.mark.unit
+def test_store_update_success():
+    store = Store()
+    key = StringKey("foo")
+    value = "bar"
+    updated_value = "baz"
+
+    store.add(key=key, value=value)
+    store.update(key=key, value=updated_value)
+
+    assert store.get(key) == updated_value
+
+
+@pytest.mark.unit
+def test_store_update_failure():
+    store = Store()
+    key = StringKey("foo")
+    value = "bar"
+
+    with pytest.raises(StoreBackendError) as e:
+        store.update(key=key, value=value)
+
+    assert "Store does not have a value associated the following key" in str(e.value)
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("previous_key_exists", [True, False])
+def test_store_add_or_update(previous_key_exists: bool):
+    store = Store()
+    key = StringKey("foo")
+    value = "bar"
+
+    if previous_key_exists:
+        store.add(key=key, value=None)
+
+    store.add_or_update(key=key, value=value)
+    assert store.get(key) == value
diff --git a/tests/data_context/store/test_store_backends.py b/tests/data_context/store/test_store_backends.py
index a3f0be60e..91085af0f 100644
--- a/tests/data_context/store/test_store_backends.py
+++ b/tests/data_context/store/test_store_backends.py
@@ -1,7 +1,7 @@
 import datetime
 import json
 import os
-from unittest.mock import patch
+from unittest import mock
 
 import boto3
 import pyparsing as pp
@@ -1052,7 +1052,7 @@ def test_TupleGCSStoreBackend_base_public_path():
     project = "dummy-project"
     base_public_path = "http://www.test.com/"
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
         mock_client = mock_gcs_client.return_value
         mock_bucket = mock_client.get_bucket.return_value
         mock_blob = mock_bucket.blob.return_value
@@ -1107,7 +1107,7 @@ def test_TupleGCSStoreBackend():
     project = "dummy-project"
     base_public_path = "http://www.test.com/"
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
 
         mock_client = mock_gcs_client.return_value
         mock_bucket = mock_client.bucket.return_value
@@ -1130,7 +1130,7 @@ def test_TupleGCSStoreBackend():
             b"aaa", content_type="text/html"
         )
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
         mock_client = mock_gcs_client.return_value
         mock_bucket = mock_client.bucket.return_value
         mock_blob = mock_bucket.blob.return_value
@@ -1151,7 +1151,7 @@ def test_TupleGCSStoreBackend():
             b"aaa", content_type="image/png"
         )
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
 
         mock_client = mock_gcs_client.return_value
         mock_bucket = mock_client.bucket.return_value
@@ -1168,7 +1168,7 @@ def test_TupleGCSStoreBackend():
         mock_blob.download_as_string.assert_called_once()
         mock_str.decode.assert_called_once_with("utf-8")
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
 
         mock_client = mock_gcs_client.return_value
 
@@ -1187,7 +1187,7 @@ def test_TupleGCSStoreBackend():
         except NotFound:
             pass
 
-    with patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
+    with mock.patch("google.cloud.storage.Client", autospec=True) as mock_gcs_client:
         mock_gcs_client.side_effect = InvalidKeyError("Hi I am an InvalidKeyError")
         with pytest.raises(InvalidKeyError):
             my_store.get(("non_existent_key",))
@@ -1225,7 +1225,7 @@ def test_TupleAzureBlobStoreBackend_connection_string():
         connection_string=connection_string, prefix=prefix, container=container
     )
 
-    with patch(
+    with mock.patch(
         "azure.storage.blob.BlobServiceClient", autospec=True
     ) as mock_azure_blob_client:
 
@@ -1268,10 +1268,10 @@ def test_TupleAzureBlobStoreBackend_account_url():
         account_url=account_url, prefix=prefix, container=container
     )
 
-    with patch(
+    with mock.patch(
         "azure.storage.blob.BlobServiceClient", autospec=True
     ) as mock_azure_blob_client:
-        with patch(
+        with mock.patch(
             "azure.identity.DefaultAzureCredential", autospec=True
         ) as mock_azure_credential:
             mock_container_client = my_store._container_client
@@ -1370,7 +1370,7 @@ def test_InlineStoreBackend(empty_data_context: DataContext) -> None:
     # test valid .set
     key = DataContextVariableKey()
     tuple_ = key.to_tuple()
-    with patch(
+    with mock.patch(
         "great_expectations.data_context.store.InlineStoreBackend._save_changes"
     ) as mock_save:
         inline_store_backend.set(tuple_, new_config_version)
@@ -1572,3 +1572,65 @@ def test_InMemoryStoreBackend_build_Key() -> None:
     assert store_backend.build_key(name=name) == DataContextVariableKey(
         resource_name=name
     )
+
+
+@pytest.mark.unit
+def test_InMemoryStoreBackend_add_success():
+    store_backend = InMemoryStoreBackend()
+    key = ("foo",)
+    value = "bar"
+
+    store_backend.add(key=key, value=value)
+    assert key in store_backend.list_keys()
+
+
+@pytest.mark.unit
+def test_InMemoryStoreBackend_add_failure():
+    store_backend = InMemoryStoreBackend()
+    key = ("foo",)
+    value = "bar"
+
+    store_backend.add(key=key, value=value)
+    with pytest.raises(StoreBackendError) as e:
+        store_backend.add(key=key, value=value)
+
+    assert "Store already has the following key" in str(e.value)
+
+
+@pytest.mark.unit
+def test_InMemoryStoreBackend_update_success():
+    store_backend = InMemoryStoreBackend()
+    key = ("foo",)
+    value = "bar"
+    updated_value = "baz"
+
+    store_backend.add(key=key, value=value)
+    store_backend.update(key=key, value=updated_value)
+
+    assert store_backend.get(key) == updated_value
+
+
+@pytest.mark.unit
+def test_InMemoryStoreBackend_update_failure():
+    store_backend = InMemoryStoreBackend()
+    key = ("foo",)
+    value = "bar"
+
+    with pytest.raises(StoreBackendError) as e:
+        store_backend.update(key=key, value=value)
+
+    assert "Store does not have a value associated the following key" in str(e.value)
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("previous_key_exists", [True, False])
+def test_InMemoryStoreBackend_add_or_update(previous_key_exists: bool):
+    store_backend = InMemoryStoreBackend()
+    key = ("foo",)
+    value = "bar"
+
+    if previous_key_exists:
+        store_backend.add(key=key, value=None)
+
+    store_backend.add_or_update(key=key, value=value)
+    assert store_backend.get(key) == value
diff --git a/tests/data_context/test_data_context_state_management.py b/tests/data_context/test_data_context_state_management.py
index e95029412..4307c760c 100644
--- a/tests/data_context/test_data_context_state_management.py
+++ b/tests/data_context/test_data_context_state_management.py
@@ -30,9 +30,20 @@ class ExpectationsStoreSpy(ExpectationsStore):
         self.save_count = 0
         super().__init__()
 
-    def set(self, key, value, **kwargs):
+    def add(self, key, value, **kwargs):
+        ret = super().add(key=key, value=value, **kwargs)
         self.save_count += 1
-        return super().set(key=key, value=value, **kwargs)
+        return ret
+
+    def update(self, key, value, **kwargs):
+        ret = super().update(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
+
+    def add_or_update(self, key, value, **kwargs):
+        ret = super().add_or_update(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
 
 
 class ProfilerStoreSpy(ProfilerStore):
@@ -43,9 +54,20 @@ class ProfilerStoreSpy(ProfilerStore):
         self.save_count = 0
         super().__init__(ProfilerStoreSpy.STORE_NAME)
 
-    def set(self, key, value, **kwargs):
+    def add(self, key, value, **kwargs):
+        ret = super().add(key=key, value=value, **kwargs)
         self.save_count += 1
-        return super().set(key=key, value=value, **kwargs)
+        return ret
+
+    def update(self, key, value, **kwargs):
+        ret = super().update(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
+
+    def add_or_update(self, key, value, **kwargs):
+        ret = super().add_or_update(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
 
 
 class CheckpointStoreSpy(CheckpointStore):
@@ -56,9 +78,20 @@ class CheckpointStoreSpy(CheckpointStore):
         self.save_count = 0
         super().__init__(CheckpointStoreSpy.STORE_NAME)
 
-    def set(self, key, value, **kwargs):
+    def add(self, key, value, **kwargs):
+        ret = super().add(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
+
+    def update(self, key, value, **kwargs):
+        ret = super().update(key=key, value=value, **kwargs)
         self.save_count += 1
-        return super().set(key=key, value=value, **kwargs)
+        return ret
+
+    def add_or_update(self, key, value, **kwargs):
+        ret = super().add_or_update(key=key, value=value, **kwargs)
+        self.save_count += 1
+        return ret
 
 
 class EphemeralDataContextSpy(EphemeralDataContext):
@@ -355,11 +388,7 @@ def test_add_expectation_suite_namespace_collision_failure(
     with pytest.raises(gx_exceptions.DataContextError) as e:
         context.add_expectation_suite(expectation_suite_name=suite_name)
 
-    assert f"expectation_suite with name {suite_name} already exists" in str(e.value)
-    assert (
-        "please delete or update it using `delete_expectation_suite` or `update_expectation_suite`"
-        in str(e.value)
-    )
+    assert f"An ExpectationSuite named {suite_name} already exists" in str(e.value)
     assert context.expectations_store.save_count == 1
 
 
@@ -426,10 +455,12 @@ def test_update_expectation_suite_failure(
     suite_name = "my_brand_new_suite"
     suite = ExpectationSuite(expectation_suite_name=suite_name)
 
-    with pytest.raises(gx_exceptions.DataContextError) as e:
+    with pytest.raises(gx_exceptions.ExpectationSuiteError) as e:
         _ = context.update_expectation_suite(suite)
 
-    assert f"expectation_suite with name {suite_name} does not exist." in str(e.value)
+    assert f"Could not find an existing ExpectationSuite named {suite_name}." in str(
+        e.value
+    )
 
 
 @pytest.mark.unit
@@ -589,6 +620,31 @@ def test_add_profiler_with_existing_profiler(
     assert context.profiler_store.save_count == 1
 
 
+@pytest.mark.unit
+def test_add_profiler_namespace_collision_failure(
+    in_memory_data_context: EphemeralDataContextSpy,
+    profiler_rules: dict,
+):
+    context = in_memory_data_context
+
+    name = "my_rbp"
+    profiler = RuleBasedProfiler(
+        name=name,
+        config_version=1.0,
+        rules=profiler_rules,
+        data_context=context,
+    )
+
+    _ = context.add_profiler(profiler=profiler)
+    assert context.profiler_store.save_count == 1
+
+    with pytest.raises(gx_exceptions.ProfilerError) as e:
+        _ = context.add_profiler(profiler=profiler)
+
+    assert f"Profiler named {name} already exists" in str(e.value)
+    assert context.profiler_store.save_count == 1  # Should not have changed
+
+
 @pytest.mark.unit
 @pytest.mark.parametrize(
     "profiler,profiler_name",
@@ -643,20 +699,23 @@ def test_update_profiler_success(
 
 
 @pytest.mark.unit
-def test_update_profiler_failure(in_memory_data_context: EphemeralDataContextSpy):
+def test_update_profiler_failure(
+    in_memory_data_context: EphemeralDataContextSpy, profiler_rules: dict
+):
     context = in_memory_data_context
 
     name = "my_rbp"
     profiler = RuleBasedProfiler(
         name=name,
         config_version=1.0,
+        rules=profiler_rules,
         data_context=context,
     )
 
     with pytest.raises(gx_exceptions.ProfilerNotFoundError) as e:
         _ = context.update_profiler(profiler)
 
-    assert f"Non-existent Profiler configuration named {name}" in str(e.value)
+    assert f"Could not find an existing Profiler named {name}" in str(e.value)
 
 
 @pytest.mark.unit
@@ -786,6 +845,23 @@ def test_add_checkpoint_with_existing_checkpoint(
     assert context.checkpoint_store.save_count == 1
 
 
+@pytest.mark.unit
+def test_add_checkpoint_namespace_collision_failure(
+    in_memory_data_context: EphemeralDataContextSpy,
+):
+    context = in_memory_data_context
+    checkpoint_name = "my_checkpoint"
+
+    _ = context.add_checkpoint(name=checkpoint_name, class_name="Checkpoint")
+    assert context.checkpoint_store.save_count == 1
+
+    with pytest.raises(gx_exceptions.CheckpointError) as e:
+        _ = context.add_checkpoint(name=checkpoint_name, class_name="Checkpoint")
+
+    assert f"Checkpoint named {checkpoint_name} already exists" in str(e.value)
+    assert context.checkpoint_store.save_count == 1  # Should not have changed
+
+
 @pytest.mark.unit
 @pytest.mark.parametrize(
     "checkpoint, checkpoint_name",
@@ -871,7 +947,7 @@ def test_update_checkpoint_failure(in_memory_data_context: EphemeralDataContextS
     with pytest.raises(gx_exceptions.CheckpointNotFoundError) as e:
         context.update_checkpoint(checkpoint)
 
-    assert f"Could not find a Checkpoint named {name}" in str(e.value)
+    assert f"Could not find an existing Checkpoint named {name}" in str(e.value)
 
 
 @pytest.mark.unit
diff --git a/tests/rule_based_profiler/test_rule_based_profiler.py b/tests/rule_based_profiler/test_rule_based_profiler.py
index de458b6d1..4c84aec0c 100644
--- a/tests/rule_based_profiler/test_rule_based_profiler.py
+++ b/tests/rule_based_profiler/test_rule_based_profiler.py
@@ -1171,17 +1171,20 @@ def test_add_profiler(
     profiler_config_with_placeholder_args: RuleBasedProfilerConfig,
 ):
     mock_data_context.cloud_mode = False
+
+    profiler_args = profiler_config_with_placeholder_args.to_dict()
+    for attr in ("class_name", "module_name"):
+        profiler_args.pop(attr)
+
     profiler: RuleBasedProfiler = RuleBasedProfiler.add_profiler(
-        profiler_config_with_placeholder_args,
         data_context=mock_data_context,
         profiler_store=mock_data_context.profiler_store,
+        **profiler_args,
     )
 
     assert isinstance(profiler, RuleBasedProfiler)
     assert profiler.name == profiler_config_with_placeholder_args.name
-    assert mock_data_context.profiler_store.set.call_args == mock.call(
-        key=profiler_key, value=profiler_config_with_placeholder_args
-    )
+    mock_data_context.profiler_store.add.asset_called_once()
 
 
 @pytest.mark.cloud
@@ -1191,21 +1194,23 @@ def test_add_profiler_ge_cloud_mode(
     ge_cloud_profiler_key: GXCloudIdentifier,
     profiler_config_with_placeholder_args: RuleBasedProfilerConfig,
 ):
+    profiler_args = profiler_config_with_placeholder_args.to_dict()
+    for attr in ("class_name", "module_name"):
+        profiler_args.pop(attr)
+
     with mock.patch(
         "great_expectations.data_context.data_context.CloudDataContext",
         spec=CloudDataContext,
     ) as mock_data_context:
         profiler: RuleBasedProfiler = RuleBasedProfiler.add_profiler(
-            profiler_config_with_placeholder_args,
             data_context=mock_data_context,
             profiler_store=mock_data_context.profiler_store,
+            **profiler_args,
         )
 
     assert isinstance(profiler, RuleBasedProfiler)
     assert profiler.name == profiler_config_with_placeholder_args.name
-    assert mock_data_context.profiler_store.set.call_args == mock.call(
-        key=ge_cloud_profiler_key, value=profiler_config_with_placeholder_args
-    )
+    mock_data_context.profiler_store.add.assert_called_once()
 
 
 @mock.patch("great_expectations.data_context.data_context.AbstractDataContext")
@@ -1213,41 +1218,41 @@ def test_add_profiler_ge_cloud_mode(
 def test_add_profiler_with_batch_request_containing_batch_data_raises_error(
     mock_data_context: mock.MagicMock,
 ):
-    profiler_config = RuleBasedProfilerConfig(
-        name="my_profiler_config",
-        config_version=1.0,
-        rules={
-            "rule_1": {
-                "domain_builder": {
-                    "class_name": "TableDomainBuilder",
-                    "batch_request": {
-                        "runtime_parameters": {
-                            "batch_data": pd.DataFrame()  # Cannot be serialized in store
-                        }
-                    },
+    name = "my_profiler_config"
+    config_version = 1.0
+    rules = {
+        "rule_1": {
+            "domain_builder": {
+                "class_name": "TableDomainBuilder",
+                "batch_request": {
+                    "runtime_parameters": {
+                        "batch_data": pd.DataFrame()  # Cannot be serialized in store
+                    }
                 },
-                "parameter_builders": [
-                    {
-                        "class_name": "MetricMultiBatchParameterBuilder",
-                        "name": "my_parameter",
-                        "metric_name": "my_metric",
-                    },
-                ],
-                "expectation_configuration_builders": [
-                    {
-                        "class_name": "DefaultExpectationConfigurationBuilder",
-                        "expectation_type": "expect_column_pair_values_A_to_be_greater_than_B",
-                    },
-                ],
-            }
-        },
-    )
+            },
+            "parameter_builders": [
+                {
+                    "class_name": "MetricMultiBatchParameterBuilder",
+                    "name": "my_parameter",
+                    "metric_name": "my_metric",
+                },
+            ],
+            "expectation_configuration_builders": [
+                {
+                    "class_name": "DefaultExpectationConfigurationBuilder",
+                    "expectation_type": "expect_column_pair_values_A_to_be_greater_than_B",
+                },
+            ],
+        }
+    }
 
     with pytest.raises(InvalidConfigError) as e:
         RuleBasedProfiler.add_profiler(
-            profiler_config,
             data_context=mock_data_context,
             profiler_store=mock_data_context.profiler_store,
+            name=name,
+            config_version=config_version,
+            rules=rules,
         )
 
     assert "batch_data found in batch_request" in str(e.value)