feat(ingestion/transformer): create tag if not exist #9076
Conversation
github-actions
bot
added
the
ingestion
metadata-ingestion/src/datahub/ingestion/transformer/add_dataset_tags.py
Outdated
Show resolved
Hide resolved
maggiehays
added
the
hacktoberfest-accepted
…lab/datahub-fork into master+tag-transformer
| ) | ||
|
|
||
| for mcp in mcps: | ||
| # I am not sure if we need to update the work_unit_id. @harshal please confirm it |
There was a problem hiding this comment.
no, we don't need to update the work_unit_id
| @@ -815,7 +815,7 @@ def test_pattern_dataset_tags_transformation(mock_time): | |||
| ) | |||
| ) | |||
|
|
|||
| assert len(outputs) == 3 | |||
| assert len(outputs) == 5 | |||
There was a problem hiding this comment.
we should also be checking the actual contents of the tag creations here
| @@ -237,5 +277,10 @@ def transform( | |||
| ), | |||
| metadata=record_metadata, | |||
| ) | |||
| yield from self._handle_end_of_stream( | |||
There was a problem hiding this comment.
the indentation looks incorrect here
| if mcp.aspect is None or mcp.entityUrn is None: # to silent the lint error | ||
| continue | ||
|
|
||
| record_metadata = _update_work_unit_id( |
There was a problem hiding this comment.
this _update_work_unit_id thing is kinda strange - don't think we need it?
| self, envelope: RecordEnvelope, entity_urn: str | ||
| ) -> Iterable[RecordEnvelope]: | ||
|
|
||
| if not isinstance(self, SingleAspectTransformer): |
There was a problem hiding this comment.
let's make this work for both types of transformers
| @@ -1065,6 +1067,22 @@ def parse_sql_lineage( | |||
| default_schema=default_schema, | |||
| ) | |||
|
|
|||
| def create_tag(self, tag_name: str) -> Dict[Any, Any]: | |||
There was a problem hiding this comment.
this should return just the urn, and not a dict
| class LegacyMCETransformer(Transformer, metaclass=ABCMeta): | ||
| class HandleEndOfStreamTransformer: | ||
| def handle_end_of_stream( | ||
| self, entity_urn: str |
There was a problem hiding this comment.
this method should not take in an entity_urn. It should only be called once after all records have been processed, not once per urn
| record_metadata = envelope.metadata.copy() | ||
| record_metadata.update( | ||
| { | ||
| "workunit_id": f"txform-{simple_name}-{self.aspect_name()}" |
There was a problem hiding this comment.
actually I was wrong in my earlier comment - I think this code is still required
There was a problem hiding this comment.
:) updated the code
| ) | ||
| yield from self._handle_end_of_stream( |
There was a problem hiding this comment.
this should be after the self._mark_processed(urn) call and two level of indentation less
…lab/datahub-fork into master+tag-transformer
…lab/datahub-fork into master+tag-transformer
For the dataset create the tag if not present