change partition_key_to_vars_fn to context_to_vars_fn

[gnmjulius]

Summary & Motivation

Give more options to stream events while executing dbt assets, using "context" as a variable instead of "context.partition_key" and change the name of "partition_key_to_vars_fn" to "context_to_vars_fn" for better understanding. This change relates to this GitHub issue (#15062)

How I Tested These Changes

Create the function for context_to_vars_fn (old - partition_key_to_vars_fn)

def date_from_context(context):
    # partition_key from context
    partition_key = context.partition_key

    # all_partition from context
    all_partition = context.instance.get_dynamic_partitions(partitions_def_name="test_partition")
    index = all_partition.index(partition_key)

    dbt_vars = {
        "var_1": partition_key,
        "var_2": index
    }
    return dbt_vars

Add this function to the dbt definition:

dbt_assets = load_assets_from_dbt_project(
            project_dir=DBT_PROJECT_PATH,
            profiles_dir=DBT_PROFILES,
            select=f"tag:{source}",
            op_name=f"{dbt_ops}",
            partitions_def=test_partition,
            key_prefix=source,
            context_to_vars_fn=date_from_context,   # <-------------- add the function here
            use_build_command=True,
            display_raw_sql=True
        )

[rexledesma]

If you want more control over the execution of your dbt models, we recommend that you move over to @dbt_assets, starting in our 1.4 release of Dagster. This new API is more extensible for use cases like yours.

For example, you'll have full range to define dbt vars in your asset compute body.

Discussion Link: #14477