geoHeil
04/06/2023, 12:24 PMdbt_target_schema = os.environ.get("<<env>>", "<<default>>")
Qwame
04/06/2023, 1:00 PMnode_info_to_asset_key
argument to customize the asset names in loading dbt assetsThomas Weit
04/06/2023, 1:04 PMdef get_dbt_source_database_name(node_info: Mapping[str, Any]):
if node_info["resource_type"] == "source":
components = [node_info["database"], "BigQuery", node_info["source_name"], node_info["name"]]
else:
configured_schema = node_info["config"].get("schema")
if configured_schema is not None:
components = [configured_schema, node_info["name"]]
else:
components = [node_info["name"]]
return AssetKey(components)
dbt_assets = load_assets_from_dbt_project(
DBT_PROJECT_DIR,
DBT_PROFILES_DIR,
key_prefix=[GOOGLE_CLOUD_PROJECT, "BigQuery", DBT_DATASET],
node_info_to_asset_key=get_dbt_source_database_name
)
In this process, I modify the asset keys for source models by adding the node_info["database"]
from the sources.yml file as well the string BigQuery
to the prefix.geoHeil
04/06/2023, 1:24 PM