update typing and naming

python_modules/dagster/dagster/_core/execution/context/compute.py

@@ -135,7 +135,7 @@ def op_config(self) -> Any:
         """The parsed config specific to this op."""
 
 
-class HasExecutionProperties(ABC):
+class ContextHasExecutionProperties(ABC):
     @property
     @abstractmethod
     def execution_properties(self) -> ExecutionProperties:
@@ -161,7 +161,9 @@ def __instancecheck__(cls, instance) -> bool:
 
 
 class OpExecutionContext(
-    AbstractComputeExecutionContext, HasExecutionProperties, metaclass=OpExecutionContextMetaClass
+    AbstractComputeExecutionContext,
+    ContextHasExecutionProperties,
+    metaclass=OpExecutionContextMetaClass,
 ):
     """The ``context`` object that can be made available as the first argument to the function
     used for computing an op or asset.
@@ -589,7 +591,7 @@ def retry_number(self) -> int:
         return self._step_execution_context.previous_attempt_count
 
     def describe_op(self) -> str:
-        return self.execution_info.step_description
+        return self.execution_properties.step_description
 
     @public
     def get_mapping_key(self) -> Optional[str]:

python_modules/dagster/dagster/_core/execution/plan/compute.py

@@ -36,8 +36,7 @@
 )
 from dagster._core.events import DagsterEvent
 from dagster._core.execution.context.compute import (
-    AssetExecutionContext,
-    OpExecutionContext,
+    ContextHasExecutionProperties,
 )
 from dagster._core.execution.context.system import StepExecutionContext
 from dagster._core.system_config.objects import ResolvedRunConfig
@@ -154,7 +153,7 @@ def _yield_compute_results(
     step_context: StepExecutionContext,
     inputs: Mapping[str, Any],
     compute_fn: OpComputeFunction,
-    compute_context: Union[OpExecutionContext, AssetExecutionContext],
+    compute_context: ContextHasExecutionProperties,
 ) -> Iterator[OpOutputUnion]:
     user_event_generator = compute_fn(compute_context, inputs)
 
@@ -176,7 +175,7 @@ def _yield_compute_results(
     if inspect.isasyncgen(user_event_generator):
         user_event_generator = gen_from_async_gen(user_event_generator)
 
-    step_label = compute_context.execution_info.step_description
+    step_label = compute_context.execution_properties.step_description
 
     for event in iterate_with_context(
         lambda: op_execution_error_boundary(
@@ -189,19 +188,19 @@ def _yield_compute_results(
         ),
         user_event_generator,
     ):
-        if compute_context.execution_info.op_execution_context.has_events():
-            yield from compute_context.execution_info.op_execution_context.consume_events()
+        if compute_context.execution_properties.op_execution_context.has_events():
+            yield from compute_context.execution_properties.op_execution_context.consume_events()
         yield _validate_event(event, step_context)
 
-    if compute_context.execution_info.op_execution_context.has_events():
+    if compute_context.execution_properties.op_execution_context.has_events():
         yield from compute_context.consume_events()
 
 
 def execute_core_compute(
     step_context: StepExecutionContext,
     inputs: Mapping[str, Any],
     compute_fn: OpComputeFunction,
-    compute_context: Union[OpExecutionContext, AssetExecutionContext],
+    compute_context: ContextHasExecutionProperties,
 ) -> Iterator[OpOutputUnion]:
     """Execute the user-specified compute for the op. Wrap in an error boundary and do
     all relevant logging and metrics tracking.

python_modules/dagster/dagster/_core/execution/plan/compute_generator.py

@@ -36,7 +36,10 @@
 from dagster._utils import is_named_tuple_instance
 from dagster._utils.warnings import disable_dagster_warnings
 
-from ..context.compute import AssetExecutionContext, OpExecutionContext
+from ..context.compute import (
+    ContextHasExecutionProperties,
+    OpExecutionContext,
+)
 
 
 def create_op_compute_wrapper(
@@ -137,7 +140,7 @@ def _coerce_op_compute_fn_to_iterator(
 
 def _zip_and_iterate_op_result(
     result: Any,
-    context: Union[OpExecutionContext, AssetExecutionContext],
+    context: ContextHasExecutionProperties,
     output_defs: Sequence[OutputDefinition],
 ) -> Iterator[Tuple[int, Any, OutputDefinition]]:
     # Filtering the expected output defs here is an unfortunate temporary solution to deal with the
@@ -166,7 +169,7 @@ def _zip_and_iterate_op_result(
 # MaterializeResult.
 def _filter_expected_output_defs(
     result: Any,
-    context: Union[OpExecutionContext, AssetExecutionContext],
+    context: ContextHasExecutionProperties,
     output_defs: Sequence[OutputDefinition],
 ) -> Sequence[OutputDefinition]:
     result_tuple = (
@@ -184,7 +187,7 @@ def _filter_expected_output_defs(
 
 
 def _validate_multi_return(
-    context: Union[OpExecutionContext, AssetExecutionContext],
+    context: ContextHasExecutionProperties,
     result: Any,
     output_defs: Sequence[OutputDefinition],
 ) -> Any:
@@ -200,7 +203,7 @@ def _validate_multi_return(
     # When returning from an op with multiple outputs, the returned object must be a tuple of the same length as the number of outputs. At the time of the op's construction, we verify that a provided annotation is a tuple with the same length as the number of outputs, so if the result matches the number of output defs on the op, it will transitively also match the annotation.
     if not isinstance(result, tuple):
         raise DagsterInvariantViolationError(
-            f"{context.execution_info.step_description} has multiple outputs, but only one "
+            f"{context.execution_properties.step_description} has multiple outputs, but only one "
             f"output was returned of type {type(result)}. When using "
             "multiple outputs, either yield each output, or return a tuple "
             "containing a value for each output. Check out the "
@@ -211,9 +214,9 @@ def _validate_multi_return(
     if not len(output_tuple) == len(output_defs):
         raise DagsterInvariantViolationError(
             "Length mismatch between returned tuple of outputs and number of "
-            f"output defs on {context.execution_info.step_description}. Output tuple has "
+            f"output defs on {context.execution_properties.step_description}. Output tuple has "
             f"{len(output_tuple)} outputs, while "
-            f"{context.execution_info.step_description} has {len(output_defs)} outputs."
+            f"{context.execution_properties.step_description} has {len(output_defs)} outputs."
         )
     return result
 
@@ -247,7 +250,7 @@ def _check_output_object_name(
 
 def validate_and_coerce_op_result_to_iterator(
     result: Any,
-    context: Union[AssetExecutionContext, OpExecutionContext],
+    context: ContextHasExecutionProperties,
     output_defs: Sequence[OutputDefinition],
 ) -> Iterator[Any]:
     if inspect.isgenerator(result):
@@ -256,7 +259,7 @@ def validate_and_coerce_op_result_to_iterator(
             yield event
     elif isinstance(result, (AssetMaterialization, ExpectationResult)):
         raise DagsterInvariantViolationError(
-            f"Error in {context.execution_info.step_description}: If you are "
+            f"Error in {context.execution_properties.step_description}: If you are "
             "returning an AssetMaterialization "
             "or an ExpectationResult from "
             "an op you must yield them "
@@ -269,8 +272,8 @@ def validate_and_coerce_op_result_to_iterator(
         yield result
     elif result is not None and not output_defs:
         raise DagsterInvariantViolationError(
-            f"Error in {context.execution_info.step_description}: Unexpectedly returned output of type"
-            f" {type(result)}. {context.execution_info.step_description} is explicitly defined to"
+            f"Error in {context.execution_properties.step_description}: Unexpectedly returned output of type"
+            f" {type(result)}. {context.execution_properties.step_description} is explicitly defined to"
             " return no results."
         )
     # `requires_typed_event_stream` is a mode where we require users to return/yield exactly the
@@ -295,15 +298,15 @@ def validate_and_coerce_op_result_to_iterator(
             if output_def.is_dynamic:
                 if not isinstance(element, list):
                     raise DagsterInvariantViolationError(
-                        f"Error with output for {context.execution_info.step_description}: "
+                        f"Error with output for {context.execution_properties.step_description}: "
                         f"dynamic output '{output_def.name}' expected a list of "
                         "DynamicOutput objects, but instead received instead an "
                         f"object of type {type(element)}."
                     )
                 for item in element:
                     if not isinstance(item, DynamicOutput):
                         raise DagsterInvariantViolationError(
-                            f"Error with output for {context.execution_info.step_description}: "
+                            f"Error with output for {context.execution_properties.step_description}: "
                             f"dynamic output '{output_def.name}' at position {position} expected a "
                             "list of DynamicOutput objects, but received an "
                             f"item with type {type(item)}."
@@ -325,7 +328,7 @@ def validate_and_coerce_op_result_to_iterator(
                     annotation
                 ):
                     raise DagsterInvariantViolationError(
-                        f"Error with output for {context.execution_info.step_description}: received Output object for"
+                        f"Error with output for {context.execution_properties.step_description}: received Output object for"
                         f" output '{output_def.name}' which does not have an Output annotation."
                         f" Annotation has type {annotation}."
                     )
@@ -343,15 +346,15 @@ def validate_and_coerce_op_result_to_iterator(
                 # output object was not received, throw an error.
                 if is_generic_output_annotation(annotation):
                     raise DagsterInvariantViolationError(
-                        f"Error with output for {context.execution_info.step_description}: output "
+                        f"Error with output for {context.execution_properties.step_description}: output "
                         f"'{output_def.name}' has generic output annotation, "
                         "but did not receive an Output object for this output. "
                         f"Received instead an object of type {type(element)}."
                     )
                 if result is None and output_def.is_required is False:
                     context.log.warning(
                         'Value "None" returned for non-required output '
-                        f'"{output_def.name}" of {context.execution_info.step_description}. '
+                        f'"{output_def.name}" of {context.execution_properties.step_description}. '
                         "This value will be passed to downstream "
                         f"{context.op_def.node_type_str}s. For conditional "
                         "execution, results must be yielded: "

python_modules/dagster/dagster_tests/core_tests/execution_tests/test_asset_execution_context.py

@@ -79,8 +79,8 @@ def test_deprecation_warnings():
         "is_subset",
         "partition_keys",
         "get",
-        "execution_info",
-        "_execution_info",
+        "execution_properties",
+        "_execution_props",
     ]
 
     other_ignores = [