Hello ... How do you run a "dbt source freshness" with

@dbt_assets

? There are no models associated (to be materialized) when dbt runs a source freshness so is this even possible? I.e. something like below. My first attempt was just putting together a function that only ran a source freshness (the manifest.json has one model in it)

@dbt_assets(
    manifest=dbt_manifest_path,
    dagster_dbt_translator=CustomDagsterDbtTranslator(),
    io_manager_key="snowflake_io_manager",
)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["source","freshness"], context=context).stream()

Running dbt command: `dbt --no-write-json --no-use-colors source freshness --target dev --select fqn:the_models_@dbt_assets`.

Hi Jason, what is your expected outcome here? From my understanding,

dbt source freshness

creates a

sources.json

, so you can retrieve that file after your computation is completed, and then use it however you want.

@dbt_assets(
    manifest=dbt_manifest_path,
    dagster_dbt_translator=CustomDagsterDbtTranslator(),
    io_manager_key="snowflake_io_manager",
)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCliResource):
    dbt_source_freshness_invocation = dbt.cli(["source","freshness"], context=context).wait()
    
    sources_json = dbt_source_freshness_invocation.get_artifact("sources.json")

    ...

    yield from dbt.cli(["run"], context=context)

Ooooh, okay - i was thinking that any warnings or errors would be thrown up and caught by Dagster but it sounds like I'll just parse the file myself and then act on anything of interest. Thanks Rex!