Hello < rex> I m transitioning from the old to the new `dags dagster #integration-dbt

Hello <@U01AV4K3W14>, I'm transitioning from the o...

Quentin Gaborit

08/14/2023, 1:25 PM

Hello @rex, I'm transitioning from the old to the new

dagster-dbt

interface. Before the change, I used to have factories for generating

meltano

dbt

software-defined-asset, and to move to the new API I was thinking of creating a wrapper like:

Copy code

def dbt_asset_factory(
    select: str,
    io_manager_key: str = "bigquery_io_manager",
) -> list[AssetsDefinition]:
    @dbt_assets(
        manifest=Path(DBT_MANIFEST_PATH),
        io_manager_key=io_manager_key,
        select=select,
    )
    def compute(
        context: AssetExecutionContext,
        dbt_cli_resource: DbtCliResource,
    ):
        yield from dbt_cli_resource.cli(["run"], context=context).stream()
        dbt_test_invocation = dbt_cli_resource.cli(["test"], context=context)
        if not dbt_test_invocation.is_successful():
            raise Exception("my exception")

    return [compute]

Unfortunately hat snippet won't work because the

compute

function should have unique identifier to ensure the Dagster dependencies inference will work correctly. Have you ever encountered such use case? If yes do you have any idea of how to get there?

🤖 1

rex

08/14/2023, 1:29 PM

Here’s a similar question: https://dagster.slack.com/archives/C04CW71AGBW/p1690299656903019. For now, you should overwrite the name of your function, by setting

__name__

rex

08/14/2023, 1:31 PM

Something like:

Copy code

def dbt_asset_factory(
    select: str,
    io_manager_key: str = "bigquery_io_manager",
) -> list[AssetsDefinition]:
    def compute(
        context: AssetExecutionContext,
        dbt_cli_resource: DbtCliResource,
    ):
        yield from dbt_cli_resource.cli(["run"], context=context).stream()
        dbt_test_invocation = dbt_cli_resource.cli(["test"], context=context)
        if not dbt_test_invocation.is_successful():
            raise Exception("my exception")

     compute.__name__ = "set_unique_name_here"

    my_dbt_assets = dbt_assets(
        manifest=Path(DBT_MANIFEST_PATH),
        io_manager_key=io_manager_key,
        select=select,
    )(compute)

    return [my_dbt_assets]

Quentin Gaborit

08/14/2023, 1:46 PM

That is it indeed, I tried the renaming of the function but without moving the decorator

dbt_assets()

it couldn't work

Quentin Gaborit

08/14/2023, 1:46 PM

Thanks for the help dagster yay

rex

08/14/2023, 1:50 PM

Glad it worked! I can add this as an example to the

@dbt_assets

decorator API docstring for future reference

Quentin Gaborit

08/14/2023, 1:51 PM

That'd for sure help other folks that want to move as much redundant SDA code to

/resources

6 Views

Open in Slack

Previous Next