Extract compute_required_resource_keys

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

@@ -509,6 +509,19 @@ def __call__(self, fn: Callable[..., Any]) -> AssetsDefinition:
         )
 
 
+def compute_required_resource_keys(required_resource_keys, resource_defs, fn) -> Set[str]:
+    bare_required_resource_keys = required_resource_keys.copy()
+    resource_defs_keys = set(resource_defs.keys())
+    required_resource_keys = bare_required_resource_keys | resource_defs_keys
+    arg_resource_keys = {arg.name for arg in get_resource_args(fn)}
+    check.param_invariant(
+        len(bare_required_resource_keys or []) == 0 or len(arg_resource_keys) == 0,
+        "Cannot specify resource requirements in both @multi_asset decorator and as"
+        " arguments to the decorated function",
+    )
+    return required_resource_keys - arg_resource_keys
+
+
 @experimental_param(param="resource_defs")
 @deprecated_param(
     param="non_argument_deps", breaking_version="2.0.0", additional_warn_text="use `deps` instead."
@@ -646,10 +659,6 @@ def my_function(asset0):
         additional_message="Only dicts are supported for asset config_schema.",
     )
 
-    bare_required_resource_keys = required_resource_keys.copy()
-    resource_defs_keys = set(resource_defs.keys())
-    required_resource_keys = bare_required_resource_keys | resource_defs_keys
-
     asset_out_map: Mapping[str, AssetOut] = {} if outs is None else outs
 
     def inner(fn: Callable[..., Any]) -> AssetsDefinition:
@@ -690,27 +699,20 @@ def inner(fn: Callable[..., Any]) -> AssetsDefinition:
                 can_subset=can_subset,
             )
 
-        arg_resource_keys = {arg.name for arg in get_resource_args(fn)}
-        check.param_invariant(
-            len(bare_required_resource_keys or []) == 0 or len(arg_resource_keys) == 0,
-            "Cannot specify resource requirements in both @multi_asset decorator and as"
-            " arguments to the decorated function",
-        )
-
         check.invariant(
             len(in_out_mapper.overlapping_output_names) == 0,
             f"Check output names overlap with asset output names: {in_out_mapper.overlapping_output_names}",
         )
 
         with disable_dagster_warnings():
-            op_required_resource_keys = required_resource_keys - arg_resource_keys
-
             op = _Op(
                 name=op_name,
                 description=description,
                 ins=in_out_mapper.asset_ins_by_input_names,
                 out=in_out_mapper.combined_outs_by_output_name,
-                required_resource_keys=op_required_resource_keys,
+                required_resource_keys=compute_required_resource_keys(
+                    required_resource_keys, resource_defs, fn
+                ),
                 tags={
                     **({"kind": compute_kind} if compute_kind else {}),
                     **(op_tags or {}),