add DI

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

@@ -55,7 +55,7 @@
 from dagster._utils.forked_pdb import ForkedPdb
 from dagster._utils.merger import merge_dicts
 
-from .compute import OpExecutionContext
+from .compute import ExecutionInfo, OpExecutionContext
 from .system import StepExecutionContext, TypeCheckContext
 
 
@@ -125,6 +125,7 @@ def __init__(
         self._hook_defs = None
         self._tags = {}
         self._seen_outputs = {}
+        self._execution_info = None
 
         # maintain init time versions of these values so we can unbind the context
         self._init_op_config = op_config
@@ -236,6 +237,10 @@ def bind(
             raise DagsterInvalidInvocationError("Cannot provide config in both context and kwargs")
         self._op_config = resolve_bound_config(config_from_args or self.op_config, op_def)
 
+        self._execution_info = ExecutionInfo(
+            step_description=f"op {self._alias}", op_execution_context=self
+        )
+
         self._requires_typed_event_stream = False
         self._typed_event_stream_error_message = None
         self._bound = True