yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-11-02 11:39:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added dbt fixes (#10132)

Browse Source
This commit is contained in:
Onkar Ravgan 2023-02-08 17:12:19 +05:30 committed by GitHub
parent c1102b13fa
commit 6d94a03fc7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 37 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
ingestion/src/metadata/ingestion/source/database/dbt/metadata.py
View File

@ -369,9 +369,12 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
DbtCommonEnum.UPSTREAM.value DbtCommonEnum.UPSTREAM.value
] = self.parse_upstream_nodes(manifest_entities, manifest_node) ] = self.parse_upstream_nodes(manifest_entities, manifest_node)
self.context.dbt_tests[key][DbtCommonEnum.RESULTS.value] = next( self.context.dbt_tests[key][DbtCommonEnum.RESULTS.value] = next(
item (
for item in dbt_objects.dbt_run_results.results item
if item.unique_id == key for item in dbt_objects.dbt_run_results.results
if item.unique_id == key
),
None,
) )
def yield_data_models(self, dbt_objects: DbtObjects) -> Iterable[DataModelLink]: def yield_data_models(self, dbt_objects: DbtObjects) -> Iterable[DataModelLink]:
@ -418,7 +421,7 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
model_name = ( model_name = (
manifest_node.alias manifest_node.alias
if manifest_node.alias if hasattr(manifest_node, "alias") and manifest_node.alias
else manifest_node.name else manifest_node.name
) )
logger.info(f"Processing DBT node: {model_name}") logger.info(f"Processing DBT node: {model_name}")
@ -498,7 +501,9 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
try: try:
parent_node = manifest_entities[node] parent_node = manifest_entities[node]
table_name = ( table_name = (
parent_node.alias if parent_node.alias else parent_node.name parent_node.alias
if hasattr(parent_node, "alias") and parent_node.alias
else parent_node.name
) )
parent_fqn = fqn.build( parent_fqn = fqn.build(
self.metadata, self.metadata,
@ -507,7 +512,7 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
database_name=parent_node.database, database_name=parent_node.database,
schema_name=parent_node.schema_, schema_name=parent_node.schema_,
table_name=table_name, table_name=table_name,
).lower() )
if parent_fqn: if parent_fqn:
upstream_nodes.append(parent_fqn) upstream_nodes.append(parent_fqn)
except Exception as exc: # pylint: disable=broad-except except Exception as exc: # pylint: disable=broad-except
@ -536,12 +541,16 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
column_name = ( column_name = (
catalog_column.name if catalog_column else manifest_column.name catalog_column.name if catalog_column else manifest_column.name
) )
column_description = None
if catalog_column and catalog_column.comment:
column_description = catalog_column.comment
columns.append( columns.append(
Column( Column(
name=column_name, name=column_name,
description=manifest_column.description description=manifest_column.description
if manifest_column.description if manifest_column.description
else catalog_column.comment, else column_description,
dataType=ColumnTypeParser.get_column_type( dataType=ColumnTypeParser.get_column_type(
catalog_column.type catalog_column.type
if catalog_column if catalog_column
@ -863,7 +872,7 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
except Exception as err: # pylint: disable=broad-except except Exception as err: # pylint: disable=broad-except
logger.debug(traceback.format_exc()) logger.debug(traceback.format_exc())
logger.error( logger.error(
f"Failed capture tests results for node: {manifest_node.name} {err}" f"Failed to capture tests results for node: {manifest_node.name} {err}"
) )
def create_test_case_parameter_definitions(self, dbt_test): def create_test_case_parameter_definitions(self, dbt_test):

28
ingestion/src/metadata/utils/dbt_config.py
View File

@ -182,18 +182,30 @@ def _(config: DbtCloudConfig): # pylint: disable=too-many-locals
runs_data = response.get("data") runs_data = response.get("data")
if runs_data: if runs_data:
run_id = runs_data[0]["id"] run_id = runs_data[0]["id"]
logger.debug("Requesting [dbt_catalog]") try:
dbt_catalog = client.get( logger.debug("Requesting [dbt_catalog]")
f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_CATALOG_FILE_NAME}" dbt_catalog = client.get(
) f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_CATALOG_FILE_NAME}"
)
except Exception as exc:
logger.info(
f"dbt catalog file not found, skipping the catalog file: {exc}"
)
logger.debug(traceback.format_exc())
logger.debug("Requesting [dbt_manifest]") logger.debug("Requesting [dbt_manifest]")
dbt_manifest = client.get( dbt_manifest = client.get(
f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_MANIFEST_FILE_NAME}" f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_MANIFEST_FILE_NAME}"
) )
logger.debug("Requesting [dbt_run_results]") try:
dbt_run_results = client.get( logger.debug("Requesting [dbt_run_results]")
f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_RUN_RESULTS_FILE_NAME}" dbt_run_results = client.get(
) f"/accounts/{account_id}/runs/{run_id}/artifacts/{DBT_RUN_RESULTS_FILE_NAME}"
)
except Exception as exc:
logger.info(
f"dbt run_results file not found, skipping dbt tests: {exc}"
)
logger.debug(traceback.format_exc())
if not dbt_manifest: if not dbt_manifest:
raise DBTConfigException("Manifest file not found in DBT Cloud") raise DBTConfigException("Manifest file not found in DBT Cloud")