[definition-tags] Move tags normalization logic (#24791)

## Summary & Motivation

Move the two separate tags normalization routines to
`dagster._utils.tags` and add some explanatory comments.

This is preparation for merging the two routines.

## How I Tested These Changes

Existing test suite.

## Changelog

NOCHANGELOG

- [ ] `NEW` _(added new feature or capability)_
- [ ] `BUGFIX` _(fixed a bug)_
- [ ] `DOCS` _(added or updated documentation)_

python_modules/dagster-graphql/dagster_graphql/client/client.py

@@ -6,7 +6,7 @@
 from dagster import DagsterRunStatus
 from dagster._annotations import deprecated, public
 from dagster._core.definitions.run_config import RunConfig, convert_config_input
-from dagster._core.definitions.utils import normalize_tags
+from dagster._utils.tags import normalize_tags
 from gql import Client, gql
 from gql.transport import Transport
 from gql.transport.exceptions import TransportServerError

python_modules/dagster/dagster/_cli/job.py

@@ -30,7 +30,6 @@
 from dagster._core.definitions import JobDefinition
 from dagster._core.definitions.reconstruct import ReconstructableJob
 from dagster._core.definitions.selector import JobSubsetSelector
-from dagster._core.definitions.utils import normalize_tags
 from dagster._core.errors import DagsterBackfillFailedError
 from dagster._core.execution.api import execute_job
 from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
@@ -60,6 +59,7 @@
 from dagster._utils.indenting_printer import IndentingPrinter
 from dagster._utils.interrupts import capture_interrupts
 from dagster._utils.merger import merge_dicts
+from dagster._utils.tags import normalize_tags
 from dagster._utils.yaml_utils import dump_run_config_yaml, load_yaml_from_glob_list
 
 T = TypeVar("T")

python_modules/dagster/dagster/_core/definitions/asset_out.py

@@ -17,12 +17,9 @@
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.input import NoValueSentinel
 from dagster._core.definitions.output import Out
-from dagster._core.definitions.utils import (
-    DEFAULT_IO_MANAGER_KEY,
-    resolve_automation_condition,
-    validate_tags_strict,
-)
+from dagster._core.definitions.utils import DEFAULT_IO_MANAGER_KEY, resolve_automation_condition
 from dagster._core.types.dagster_type import DagsterType, resolve_dagster_type
+from dagster._utils.tags import validate_tags_strict
 from dagster._utils.warnings import disable_dagster_warnings
 
 

python_modules/dagster/dagster/_core/definitions/asset_spec.py

@@ -16,12 +16,12 @@
     resolve_automation_condition,
     validate_asset_owner,
     validate_group_name,
-    validate_tags_strict,
 )
 from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster._core.storage.tags import KIND_PREFIX
 from dagster._serdes.serdes import whitelist_for_serdes
 from dagster._utils.internal_init import IHasInternalInit
+from dagster._utils.tags import validate_tags_strict
 
 if TYPE_CHECKING:
     from dagster._core.definitions.asset_dep import AssetDep, CoercibleToAssetDep

python_modules/dagster/dagster/_core/definitions/assets.py

@@ -71,12 +71,12 @@
     DEFAULT_IO_MANAGER_KEY,
     normalize_group_name,
     validate_asset_owner,
-    validate_tags_strict,
 )
 from dagster._core.errors import DagsterInvalidDefinitionError, DagsterInvariantViolationError
 from dagster._utils import IHasInternalInit
 from dagster._utils.merger import merge_dicts
 from dagster._utils.security import non_secure_md5_hash_str
+from dagster._utils.tags import validate_tags_strict
 from dagster._utils.warnings import ExperimentalWarning, disable_dagster_warnings
 
 if TYPE_CHECKING:

python_modules/dagster/dagster/_core/definitions/automation_condition_sensor_definition.py

@@ -14,7 +14,8 @@
     SensorEvaluationContext,
     SensorType,
 )
-from dagster._core.definitions.utils import check_valid_name, normalize_tags
+from dagster._core.definitions.utils import check_valid_name
+from dagster._utils.tags import normalize_tags
 
 
 def _evaluate(sensor_def: "AutomationConditionSensorDefinition", context: SensorEvaluationContext):

python_modules/dagster/dagster/_core/definitions/composition.py

@@ -44,13 +44,14 @@
 from dagster._core.definitions.output import OutputDefinition, OutputMapping
 from dagster._core.definitions.policy import RetryPolicy
 from dagster._core.definitions.resource_definition import ResourceDefinition
-from dagster._core.definitions.utils import check_valid_name, normalize_tags
+from dagster._core.definitions.utils import check_valid_name
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     DagsterInvalidInvocationError,
     DagsterInvariantViolationError,
 )
 from dagster._utils import is_named_tuple_instance
+from dagster._utils.tags import normalize_tags
 from dagster._utils.warnings import disable_dagster_warnings
 
 if TYPE_CHECKING:

python_modules/dagster/dagster/_core/definitions/decorators/asset_decorator.py

@@ -55,11 +55,11 @@
     DEFAULT_OUTPUT,
     NoValueSentinel,
     resolve_automation_condition,
-    validate_tags_strict,
 )
 from dagster._core.errors import DagsterInvalidDefinitionError, DagsterInvariantViolationError
 from dagster._core.storage.tags import KIND_PREFIX
 from dagster._core.types.dagster_type import DagsterType
+from dagster._utils.tags import validate_tags_strict
 from dagster._utils.warnings import disable_dagster_warnings
 
 

python_modules/dagster/dagster/_core/definitions/decorators/job_decorator.py

@@ -11,7 +11,7 @@
 from dagster._core.definitions.metadata import RawMetadataValue
 from dagster._core.definitions.policy import RetryPolicy
 from dagster._core.definitions.resource_definition import ResourceDefinition
-from dagster._core.definitions.utils import normalize_tags
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions.executor_definition import ExecutorDefinition

python_modules/dagster/dagster/_core/definitions/decorators/schedule_decorator.py

@@ -19,13 +19,13 @@
 )
 from dagster._core.definitions.sensor_definition import get_context_param_name
 from dagster._core.definitions.target import ExecutableDefinition
-from dagster._core.definitions.utils import normalize_tags
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     ScheduleExecutionError,
     user_code_error_boundary,
 )
 from dagster._utils import ensure_gen
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions.asset_selection import CoercibleToAssetSelection

python_modules/dagster/dagster/_core/definitions/decorators/source_asset_decorator.py

@@ -20,7 +20,7 @@
 from dagster._core.definitions.resource_annotation import get_resource_args
 from dagster._core.definitions.resource_definition import ResourceDefinition
 from dagster._core.definitions.source_asset import SourceAsset, SourceAssetObserveFunction
-from dagster._core.definitions.utils import validate_tags_strict
+from dagster._utils.tags import validate_tags_strict
 from dagster._utils.warnings import disable_dagster_warnings
 
 

python_modules/dagster/dagster/_core/definitions/dependency.py

@@ -35,11 +35,12 @@
 )
 from dagster._core.definitions.output import OutputDefinition
 from dagster._core.definitions.policy import RetryPolicy
-from dagster._core.definitions.utils import DEFAULT_OUTPUT, normalize_tags, struct_to_string
+from dagster._core.definitions.utils import DEFAULT_OUTPUT, struct_to_string
 from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster._record import record
 from dagster._serdes.serdes import whitelist_for_serdes
 from dagster._utils import hash_collection
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions.asset_layer import AssetLayer

python_modules/dagster/dagster/_core/definitions/events.py

@@ -750,7 +750,7 @@ def __new__(cls, hook_name: str, is_skipped: Optional[bool] = None):
 
 
 def validate_asset_event_tags(tags: Optional[Mapping[str, str]]) -> Optional[Mapping[str, str]]:
-    from dagster._core.definitions.utils import validate_tag_strict
+    from dagster._utils.tags import validate_tag_strict
 
     if tags is None:
         return None

python_modules/dagster/dagster/_core/definitions/job_definition.py

@@ -63,7 +63,7 @@
     ensure_requirements_satisfied,
 )
 from dagster._core.definitions.run_request import RunRequest
-from dagster._core.definitions.utils import DEFAULT_IO_MANAGER_KEY, check_valid_name, normalize_tags
+from dagster._core.definitions.utils import DEFAULT_IO_MANAGER_KEY, check_valid_name
 from dagster._core.errors import (
     DagsterInvalidConfigError,
     DagsterInvalidDefinitionError,
@@ -82,6 +82,7 @@
 from dagster._utils import IHasInternalInit
 from dagster._utils.cached_method import cached_method
 from dagster._utils.merger import merge_dicts
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._config.snap import ConfigSchemaSnapshot

python_modules/dagster/dagster/_core/definitions/node_definition.py

@@ -14,8 +14,9 @@
 from dagster._core.definitions.configurable import NamedConfigurableDefinition
 from dagster._core.definitions.hook_definition import HookDefinition
 from dagster._core.definitions.policy import RetryPolicy
-from dagster._core.definitions.utils import check_valid_name, normalize_tags
+from dagster._core.definitions.utils import check_valid_name
 from dagster._core.errors import DagsterInvariantViolationError
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions.asset_layer import AssetLayer

python_modules/dagster/dagster/_core/definitions/partition.py

@@ -32,7 +32,6 @@
 )
 from dagster._core.definitions.partition_key_range import PartitionKeyRange
 from dagster._core.definitions.run_config import RunConfig, convert_config_input
-from dagster._core.definitions.utils import normalize_tags
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     DagsterInvalidDeserializationVersionError,
@@ -44,6 +43,7 @@
 from dagster._serdes import whitelist_for_serdes
 from dagster._utils import xor
 from dagster._utils.cached_method import cached_method
+from dagster._utils.tags import normalize_tags
 from dagster._utils.warnings import normalize_renamed_param
 
 DEFAULT_DATE_FORMAT = "%Y-%m-%d"

python_modules/dagster/dagster/_core/definitions/run_request.py

@@ -30,7 +30,6 @@
 )
 from dagster._core.definitions.events import AssetKey, AssetMaterialization, AssetObservation
 from dagster._core.definitions.partition_key_range import PartitionKeyRange
-from dagster._core.definitions.utils import normalize_tags
 from dagster._core.instance import DynamicPartitionsStore
 from dagster._core.storage.dagster_run import DagsterRun, DagsterRunStatus
 from dagster._core.storage.tags import (
@@ -42,6 +41,7 @@
 from dagster._serdes.serdes import whitelist_for_serdes
 from dagster._utils.cached_method import cached_method
 from dagster._utils.error import SerializableErrorInfo
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions.job_definition import JobDefinition

python_modules/dagster/dagster/_core/definitions/schedule_definition.py

@@ -40,7 +40,7 @@
     ExecutableDefinition,
 )
 from dagster._core.definitions.unresolved_asset_job_definition import UnresolvedAssetJobDefinition
-from dagster._core.definitions.utils import check_valid_name, normalize_tags
+from dagster._core.definitions.utils import check_valid_name
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     DagsterInvalidInvocationError,
@@ -56,6 +56,7 @@
 from dagster._utils import IHasInternalInit, ensure_gen
 from dagster._utils.merger import merge_dicts
 from dagster._utils.schedules import has_out_of_range_cron_interval, is_valid_cron_schedule
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster import ResourceDefinition

python_modules/dagster/dagster/_core/definitions/sensor_definition.py

@@ -62,7 +62,7 @@
     AutomationTarget,
     ExecutableDefinition,
 )
-from dagster._core.definitions.utils import check_valid_name, normalize_tags
+from dagster._core.definitions.utils import check_valid_name
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     DagsterInvalidInvocationError,
@@ -76,6 +76,7 @@
 from dagster._time import get_current_datetime
 from dagster._utils import IHasInternalInit, normalize_to_repository
 from dagster._utils.merger import merge_dicts
+from dagster._utils.tags import normalize_tags
 from dagster._utils.warnings import deprecation_warning, normalize_renamed_param
 
 if TYPE_CHECKING:

python_modules/dagster/dagster/_core/definitions/source_asset.py

@@ -45,7 +45,6 @@
     DEFAULT_GROUP_NAME,
     DEFAULT_IO_MANAGER_KEY,
     normalize_group_name,
-    validate_tags_strict,
 )
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
@@ -54,6 +53,7 @@
     DagsterInvariantViolationError,
 )
 from dagster._utils.internal_init import IHasInternalInit
+from dagster._utils.tags import validate_tags_strict
 
 if TYPE_CHECKING:
     from dagster._core.definitions.decorators.op_decorator import DecoratedOpFunction

python_modules/dagster/dagster/_core/definitions/unresolved_asset_job_definition.py

@@ -17,9 +17,9 @@
 from dagster._core.definitions.policy import RetryPolicy
 from dagster._core.definitions.resource_definition import ResourceDefinition
 from dagster._core.definitions.run_request import RunRequest
-from dagster._core.definitions.utils import normalize_tags
 from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster._core.instance import DynamicPartitionsStore
+from dagster._utils.tags import normalize_tags
 
 if TYPE_CHECKING:
     from dagster._core.definitions import JobDefinition