https://dagster.io/ logo
#ask-ai
Title
# ask-ai
s

Santosh Sewlal

02/06/2024, 8:17 PM
@Scout Please help with this error. This code
Copy code
@dbt_assets(manifest=DBT_MANIFEST_PATH)
def dag_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from load_assets_from_dbt_project(DBT_PROJECT_PATH, profiles_dir=DBT_PROFILES, use_build_command=True)
fails with
Copy code
dagster._core.errors.DagsterInvariantViolationError: Compute function for op "dag_dbt_assets" yielded a value of type <class 'dagster._core.definitions.assets.AssetsDefinition'> rather than an instance of Output, AssetMaterialization, or ExpectationResult. Values yielded by ops must be wrapped in one of these types. If your op has a single output and yields no other events, you may want to use `return` instead of `yield` in the body of your op compute function. If you are already using `return`, and you expected to return a value of type <class 'dagster._core.definitions.assets.AssetsDefinition'>, you may be inadvertently returning a generator rather than the value 

  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_plan.py", line 273, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 434, in core_dagster_event_sequence_for_step
    for user_event in check.generator(
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 117, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 87, in _asset_check_results_to_outputs_and_evaluations
    for user_event in user_event_sequence:
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/compute.py", line 198, in execute_core_compute
    for step_output in _yield_compute_results(step_context, inputs, compute_fn):
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/compute.py", line 180, in _yield_compute_results
    yield _validate_event(event, step_context)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/dagster/_core/execution/plan/compute.py", line 105, in _validate_event
    raise DagsterInvariantViolationError(
s

Scout

02/06/2024, 8:17 PM
The error message you're encountering indicates that the compute function for the op
dag_dbt_assets
is yielding an object of type
AssetsDefinition
instead of an
Output
,
AssetMaterialization
, or
ExpectationResult
. In Dagster, when you define a compute function for an op, you are expected to yield or return one of these specific types to indicate the result of the computation. The
@dbt_assets
decorator is used to define a collection of assets that are produced by dbt models. The compute function within this decorator should yield events such as
Output
,
AssetMaterialization
, or
AssetObservation
. However, the function
load_assets_from_dbt_project
seems to be returning an
AssetsDefinition
object, which is not a valid type to yield from an op compute function. To resolve this issue, you should ensure that the body of the
dag_dbt_assets
function yields the correct types. If
load_assets_from_dbt_project
is intended to define assets, you might need to refactor your code to use the assets it defines directly, rather than trying to yield them from within the
@dbt_assets
decorated function. If
load_assets_from_dbt_project
is a function that is supposed to run dbt commands and yield asset-related events, then you need to make sure that it yields the correct types of events that Dagster expects. If it's not designed to be used within an op compute function, you may need to use a different approach to integrate dbt with your Dagster assets. Without more context on what
load_assets_from_dbt_project
does, it's difficult to provide a precise solution. However, the key takeaway is that within the body of an op compute function, you should only yield or return the types of events that Dagster expects, and
AssetsDefinition
is not one of them.
s

Santosh Sewlal

02/06/2024, 8:19 PM
@Scout
from dagster_dbt import load_assets_from_dbt_project
s

Scout

02/06/2024, 8:19 PM
The function
load_assets_from_dbt_project
from the
dagster_dbt
library is designed to load a set of dbt models from a dbt project into Dagster assets. When you use this function, it creates one Dagster asset for each dbt model, and all assets will be re-materialized using a single dbt run or dbt build command. The error you're encountering suggests that there might be a misunderstanding about how to use the
load_assets_from_dbt_project
function within the Dagster framework. The
@dbt_assets
decorator is used to define assets based on dbt models, and within the decorated function, you should yield events that Dagster can handle, such as
Output
,
AssetMaterialization
, or
ExpectationResult
. If you're using
load_assets_from_dbt_project
to define assets, you should not need to use the
@dbt_assets
decorator. Instead, you would typically call
load_assets_from_dbt_project
at the module level, not within an op or asset compute function, to define your assets based on your dbt project. Here's a simplified example of how you might use `load_assets_from_dbt_project`:
Copy code
python from dagster_dbt import load_assets_from_dbt_project  # Assuming DBT_PROJECT_PATH and DBT_PROFILES_DIR are defined appropriately dbt_assets = load_assets_from_dbt_project(     project_dir=DBT_PROJECT_PATH,     profiles_dir=DBT_PROFILES_DIR,     use_build_command=True )
In this example,
dbt_assets
would be a collection of Dagster assets generated from your dbt project. You would then include these assets in your Dagster repository. Please ensure that you are following the latest documentation and examples for the
dagster_dbt
integration to correctly set up your dbt assets within Dagster. If you continue to encounter issues, you may want to consult the documentation for
dagster_dbt
or seek further assistance with a more detailed code example and context.
s

Santosh Sewlal

02/06/2024, 8:27 PM
@Scout i think the error originates at this code:
Copy code
defs = Definitions(
    assets=[dag_dbt_assets, mean_transactions_per_month],
    resources=model_resources, 
    jobs=[run_dbt_test_job]
)
can it be refactored?
s

Scout

02/06/2024, 8:27 PM
Based on the documentation, the
Definitions
class is used to collect and define assets, resources, and jobs that you want to include in your Dagster repository. However, the
dag_dbt_assets
function you've shown earlier is decorated with
@dbt_assets
, which means it's already defining assets. You should not need to include it in the
assets
list within
Definitions
. Instead, the
dag_dbt_assets
function should automatically create assets based on your dbt project when it's invoked. These assets are instances of
AssetsDefinition
, and they should not be manually added to the
Definitions
object. Here's how you might refactor your code:
Copy code
python from dagster import define_asset_job, repository from dagster_dbt import dbt_cli_resource  # Assuming mean_transactions_per_month is a valid asset definition from your_assets_module import mean_transactions_per_month  # Define your resources, including the dbt_cli_resource model_resources = {     "dbt": dbt_cli_resource.configured({...})  # Configure with appropriate settings }  # Define a job that includes the dbt assets and any other assets you have @define_asset_job(     name="run_dbt_test_job",     selection=[mean_transactions_per_month.name],  # Select specific assets if needed     resource_defs=model_resources ) def run_dbt_test_job():     # Job definition goes here     pass  # Define your repository @repository def my_repository():     return [run_dbt_test_job, mean_transactions_per_month]
In this refactored code: - We define
model_resources
with the necessary resources, including
dbt_cli_resource
. - We create a job
run_dbt_test_job
that includes the necessary assets. If
mean_transactions_per_month
is an asset, it can be included in the job's selection. - We use the
@repository
decorator to define the repository, which includes the job and the
mean_transactions_per_month
asset. Please note that the
@define_asset_job
decorator is used to create a job that materializes a set of assets. If
mean_transactions_per_month
is an asset, it can be included directly in the job definition. Make sure to adjust the code to fit the actual structure and requirements of your project. If
mean_transactions_per_month
is not an asset, you would need to adjust the selection accordingly. The
dbt_cli_resource
should be configured with the appropriate settings for your dbt project and profiles directory.
2 Views