Given this code

@op(out=Out(Optional[ModelAssemblyResults]))
def electrothermal_model_phase_group_op(
        db: DagsterDbResource,
        config: DagsterElectrothermalModelRunConfig):
    """Create single and consensus electrothermal models for the phase/group(s)
    pair.
    """
    model_assembly_results: Optional[ModelAssemblyResults] = \
        build_artifacts_and_electrothermal_models(
        run_config=get_run_config_from_dagster_run_config(config),
        db_resource=db.get_db_resource()
    )
    return Output(model_assembly_results)

@job(config=build_and_validate_electrothermal_models_config)
def build_and_validate_electrothermal_models_job():
    model_assembly_results: Optional[ModelAssemblyResults] = \
        electrothermal_model_phase_group_op()
    electrothermal_model_selection_config = \
        DagsterElectrothermalModelSelectionConfig(
            single_cell_model_ids=model_assembly_results.single_cell_model_ids,
            consensus_model_ids=model_assembly_results.consensus_model_ids
        )
    run_electrothermal_validation_sim_set(
        electrothermal_model_selection_config=electrothermal_model_selection_config
    )

How do I fix the 'InvokedNodeOutputHandle' object has no attribute error on single_cell_model_ids=model_assembly_results.single_cell_model_ids,

The error you're encountering,

'InvokedNodeOutputHandle' object has no attribute

, suggests that you're trying to access attributes on an object that is not the expected type. In Dagster, when you invoke an op within a job, the return value is not the actual output of the op, but rather an

InvokedNodeOutputHandle

. This handle is used internally by Dagster to wire up dependencies and does not have the attributes of the actual output object. To fix this issue, you need to ensure that you're not trying to directly access the attributes of the op's output within the job definition. Instead, you should pass the output of the op to another op that consumes it. Here's how you can refactor your code:

python
@op(out=Out(Optional[ModelAssemblyResults]))
def electrothermal_model_phase_group_op(
        db: DagsterDbResource,
        config: DagsterElectrothermalModelRunConfig):
    # Your existing implementation
    ...

@op
def run_electrothermal_validation_sim_set_op(
        model_assembly_results: ModelAssemblyResults):
    electrothermal_model_selection_config = \
        DagsterElectrothermalModelSelectionConfig(
            single_cell_model_ids=model_assembly_results.single_cell_model_ids,
            consensus_model_ids=model_assembly_results.consensus_model_ids
        )
    # Call the function that runs the validation simulation set
    run_electrothermal_validation_sim_set(
        electrothermal_model_selection_config=electrothermal_model_selection_config
    )

@job(config=build_and_validate_electrothermal_models_config)
def build_and_validate_electrothermal_models_job():
    model_assembly_results = electrothermal_model_phase_group_op()
    run_electrothermal_validation_sim_set_op(model_assembly_results)

In this refactored code,

run_electrothermal_validation_sim_set_op

is an op that takes the output of

electrothermal_model_phase_group_op

as an input. The job definition wires these two ops together, passing the output of the first op directly into the second op without trying to access its attributes in the job scope. This approach adheres to Dagster's data flow model, where data is passed between ops through their inputs and outputs, rather than being accessed directly in the job scope. dagster-dbt is not respecting`--indirect... button dbt_build_op does not properly handle ev... button DBT+Dagster: Seed Lineage button "__ASSET_JOB_1 has no op named" error wi... button Thank you for your feedback!