-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' of https://github.com/DalgoT4D/dbt-automation int…
…o 107-generic-function-operation
- Loading branch information
Showing
9 changed files
with
544 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| {# | ||
| Pivot values from columns to rows. Similar to pandas DataFrame melt() function. | ||
|
|
||
| Example Usage: {{ unpivot(relation=ref('users'), cast_to='integer', exclude=['id','created_at']) }} | ||
|
|
||
| Arguments: | ||
| relation: Relation object, required. | ||
| cast_to: The datatype to cast all unpivoted columns to. Default is varchar. | ||
| exclude: A list of columns to keep but exclude from the unpivot operation. Default is none. | ||
| remove: A list of columns to remove from the resulting table. Default is none. | ||
| field_name: Destination table column name for the source table column names. | ||
| value_name: Destination table column name for the pivoted values | ||
| #} | ||
|
|
||
| {% macro unpivot(relation=none, cast_to='varchar', exclude=none, remove=none, field_name='field_name', value_name='value', quote_identifiers=True) -%} | ||
| {{ return(adapter.dispatch('unpivot', 'dbt_utils')(relation, cast_to, exclude, remove, field_name, value_name, quote_identifiers)) }} | ||
| {% endmacro %} | ||
|
|
||
| {% macro default__unpivot(relation=none, cast_to='varchar', exclude=none, remove=none, field_name='field_name', value_name='value', quote_identifiers=True) -%} | ||
|
|
||
| {% if not relation %} | ||
| {{ exceptions.raise_compiler_error("Error: argument `relation` is required for `unpivot` macro.") }} | ||
| {% endif %} | ||
|
|
||
| {%- set exclude = exclude if exclude is not none else [] %} | ||
| {%- set remove = remove if remove is not none else [] %} | ||
|
|
||
| {%- set include_cols = [] %} | ||
|
|
||
| {%- set table_columns = {} %} | ||
|
|
||
| {%- do table_columns.update({relation: []}) %} | ||
|
|
||
| {%- do dbt_utils._is_relation(relation, 'unpivot') -%} | ||
| {%- do dbt_utils._is_ephemeral(relation, 'unpivot') -%} | ||
| {%- set cols = adapter.get_columns_in_relation(relation) %} | ||
|
|
||
| {%- for col in cols -%} | ||
| {%- if col.column.lower() not in remove|map('lower') and col.column.lower() not in exclude|map('lower') -%} | ||
| {% do include_cols.append(col) %} | ||
| {%- endif %} | ||
| {%- endfor %} | ||
|
|
||
|
|
||
| {%- for col in include_cols -%} | ||
| {%- set current_col_name = adapter.quote(col.column) if quote_identifiers else col.column -%} | ||
| select | ||
| {%- for exclude_col in exclude %} | ||
| {{ adapter.quote(exclude_col) if quote_identifiers else exclude_col }}, | ||
| {%- endfor %} | ||
|
|
||
| cast('{{ col.column }}' as {{ dbt.type_string() }}) as {{ adapter.quote(field_name) if quote_identifiers else field_name }}, | ||
| cast( {% if col.data_type == 'boolean' %} | ||
| {{ dbt.cast_bool_to_text(current_col_name) }} | ||
| {% else %} | ||
| {{ current_col_name }} | ||
| {% endif %} | ||
| as {{ cast_to }}) as {{ adapter.quote(value_name) if quote_identifiers else value_name }} | ||
|
|
||
| from {{ relation }} | ||
|
|
||
| {% if not loop.last -%} | ||
| union all | ||
| {% endif -%} | ||
| {%- endfor -%} | ||
|
|
||
| {%- endmacro %} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| """ | ||
| Generates a dbt model for pivot | ||
| """ | ||
|
|
||
| from logging import basicConfig, getLogger, INFO | ||
|
|
||
| from dbt_automation.utils.dbtproject import dbtProject | ||
| from dbt_automation.utils.columnutils import quote_columnname | ||
| from dbt_automation.utils.interfaces.warehouse_interface import WarehouseInterface | ||
| from dbt_automation.utils.columnutils import quote_columnname, quote_constvalue | ||
| from dbt_automation.utils.tableutils import source_or_ref | ||
|
|
||
| basicConfig(level=INFO) | ||
| logger = getLogger() | ||
|
|
||
|
|
||
| def select_from(input_table: dict): | ||
| """generates the correct FROM clause for the input table""" | ||
| selectfrom = source_or_ref(**input_table) | ||
| if input_table["input_type"] == "cte": | ||
| return f"FROM {selectfrom}\n" | ||
| return f"FROM {{{{{selectfrom}}}}}\n" | ||
|
|
||
|
|
||
| # pylint:disable=unused-argument,logging-fstring-interpolation | ||
| def pivot_dbt_sql( | ||
| config: dict, | ||
| warehouse: WarehouseInterface, | ||
| ): | ||
| """ | ||
| Generate SQL code for the coalesce_columns operation. | ||
| """ | ||
| source_columns = config.get("source_columns", []) | ||
| pivot_column_values = config.get("pivot_column_values", []) | ||
| pivot_column_name = config.get("pivot_column_name", None) | ||
| input_table = config["input"] | ||
|
|
||
| if not pivot_column_name: | ||
| raise ValueError("Pivot column name not provided") | ||
|
|
||
| dbt_code = "SELECT\n" | ||
|
|
||
| if len(source_columns) > 0: | ||
| dbt_code += ",\n".join( | ||
| [quote_columnname(col_name, warehouse.name) for col_name in source_columns] | ||
| ) | ||
| dbt_code += ",\n" | ||
|
|
||
| dbt_code += "{{ dbt_utils.pivot(" | ||
| dbt_code += quote_constvalue( | ||
| quote_columnname(pivot_column_name, warehouse.name), warehouse.name | ||
| ) | ||
| dbt_code += ", " | ||
| dbt_code += ( | ||
| "[" | ||
| + ",".join( | ||
| [ | ||
| quote_constvalue(pivot_val, warehouse.name) | ||
| for pivot_val in pivot_column_values | ||
| ] | ||
| ) | ||
| + "]" | ||
| ) | ||
| dbt_code += ")}}\n" | ||
|
|
||
| dbt_code += select_from(input_table) | ||
| if len(source_columns) > 0: | ||
| dbt_code += "GROUP BY " | ||
| dbt_code += ",".join( | ||
| [quote_columnname(col_name, warehouse.name) for col_name in source_columns] | ||
| ) | ||
|
|
||
| return dbt_code, source_columns + pivot_column_values | ||
|
|
||
|
|
||
| def pivot(config: dict, warehouse: WarehouseInterface, project_dir: str): | ||
| """ | ||
| Perform coalescing of columns and generate a DBT model. | ||
| """ | ||
| dbt_sql = "" | ||
| if config["input"]["input_type"] != "cte": | ||
| dbt_sql = ( | ||
| "{{ config(materialized='table', schema='" + config["dest_schema"] + "') }}" | ||
| ) | ||
|
|
||
| select_statement, output_cols = pivot_dbt_sql(config, warehouse) | ||
| dbt_sql += "\n" + select_statement | ||
|
|
||
| dbt_project = dbtProject(project_dir) | ||
| dbt_project.ensure_models_dir(config["dest_schema"]) | ||
|
|
||
| output_name = config["output_name"] | ||
| dest_schema = config["dest_schema"] | ||
| model_sql_path = dbt_project.write_model(dest_schema, output_name, dbt_sql) | ||
|
|
||
| return model_sql_path, output_cols |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.