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

Skipping analysis nodes (#8812)

Browse Source
This commit is contained in:
Onkar Ravgan 2022-11-17 13:54:51 +05:30 committed by GitHub
parent b74fb82c15
commit c56004c395
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 21 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
ingestion/src/metadata/ingestion/source/database/dbt_source.py
View File

@ -75,30 +75,31 @@ class DBTMixin:
"""
Returns dbt owner
"""
dbt_owner = mnode["meta"].get("owner") or cnode["metadata"].get("owner")
owner = None
if dbt_owner:
owner_name = f"*{dbt_owner}*"
user_owner_fqn = fqn.build(
self.metadata, entity_type=User, user_name=owner_name
)
if user_owner_fqn:
owner = self.metadata.get_entity_reference(
entity=User, fqn=user_owner_fqn
if mnode and cnode:
dbt_owner = mnode["meta"].get("owner") or cnode["metadata"].get("owner")
if dbt_owner:
owner_name = f"*{dbt_owner}*"
user_owner_fqn = fqn.build(
self.metadata, entity_type=User, user_name=owner_name
)
else:
team_owner_fqn = fqn.build(
self.metadata, entity_type=Team, team_name=owner_name
)
if team_owner_fqn:
if user_owner_fqn:
owner = self.metadata.get_entity_reference(
entity=Team, fqn=team_owner_fqn
entity=User, fqn=user_owner_fqn
)
else:
logger.warning(
"Unable to ingest owner from DBT since no user or"
f" team was found with name {dbt_owner}"
team_owner_fqn = fqn.build(
self.metadata, entity_type=Team, team_name=owner_name
)
if team_owner_fqn:
owner = self.metadata.get_entity_reference(
entity=Team, fqn=team_owner_fqn
)
else:
logger.warning(
"Unable to ingest owner from DBT since no user or"
f" team was found with name {dbt_owner}"
)
return owner
def _parse_data_model(self):
@ -134,6 +135,8 @@ class DBTMixin:
if mnode["resource_type"] == "test":
self.dbt_tests[key] = mnode
continue
if mnode["resource_type"] == "analysis":
continue
upstream_nodes = self._parse_data_model_upstream(mnode)
database = mnode["database"] if mnode["database"] else "default"
schema = mnode["schema"] if mnode["schema"] else "default"