ad-m-ss commented on code in PR #41422:
URL: https://github.com/apache/airflow/pull/41422#discussion_r1715438128
##########
airflow/providers/amazon/aws/transfers/s3_to_redshift.py:
##########
@@ -197,3 +199,77 @@ def execute(self, context: Context) -> None:
else:
redshift_hook.run(sql, autocommit=self.autocommit)
self.log.info("COPY command complete...")
+
+ def get_openlineage_facets_on_complete(self, task_instance):
+ """Implement on_complete as we will query destination table."""
+ from pathlib import Path
+
+ from airflow.providers.amazon.aws.utils.open_lineage import (
+ get_facets_from_redshift_table,
+ get_identity_column_lineage_facet,
+ )
+ from airflow.providers.common.compat.openlineage.facet import (
+ Dataset,
+ Identifier,
+ LifecycleStateChange,
+ LifecycleStateChangeDatasetFacet,
+ SymlinksDatasetFacet,
+ )
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ redshift_hook = self._create_hook()
+ if isinstance(redshift_hook, RedshiftDataHook):
Review Comment:
I am not sure about style guidelines of the project, but I suggest
```suggestion
if self.redshift_data_api_kwargs:
```
`isinstance` is generally discouraged in Python.
We can also `property` like:
```
@property
def use_redshift_data(self):
return bool(self.redshift_data_api_kwargs)
```
to avoid duplication of that condition in an explicit way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
