You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mobuchowski (via GitHub)" <gi...@apache.org> on 2023/05/17 15:14:43 UTC
[GitHub] [airflow] mobuchowski opened a new pull request, #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
mobuchowski opened a new pull request, #31360:
URL: https://github.com/apache/airflow/pull/31360
This PR adds OpenLineage metadata support for SFTPOperator.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228405037
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
+ try:
+ remote_host = socket.gethostbyname(remote_host)
+ except Exception as e:
Review Comment:
Yes. There are several possible socket exceptions here, but all are subclassing `OSError` with which I replaced `Exception`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228172851
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
Review Comment:
```suggestion
hook = self.sftp_hook or self.ssh_hook or SFTPHook(ssh_conn_id=self.ssh_conn_id)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1227644081
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
+ try:
+ remote_host = socket.gethostbyname(remote_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve remote hostname. Using the provided hostname without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if hasattr(hook, "port"):
+ remote_port = hook.port
+ elif hasattr(hook, "ssh_hook"):
+ remote_port = hook.ssh_hook.port
+
+ # Since v4.1.0, SFTPOperator accepts both a string (single file) and a list of
+ # strings (multiple files) as local_filepath and remote_filepath, and internally
+ # keeps them as list in both cases. But before 4.1.0, only single string is
+ # allowed. So we consider both cases here for backward compatibility.
+ if isinstance(self.local_filepath, str):
+ local_filepath = [self.local_filepath]
+ else:
+ local_filepath = self.local_filepath
+ if isinstance(self.remote_filepath, str):
+ remote_filepath = [self.remote_filepath]
+ else:
+ remote_filepath = self.remote_filepath
+
+ local_datasets = [
+ Dataset(namespace=self._get_namespace(scheme, local_host, None, path), name=path)
+ for path in local_filepath
+ ]
+ remote_datasets = [
+ Dataset(namespace=self._get_namespace(scheme, remote_host, remote_port, path), name=path)
+ for path in remote_filepath
+ ]
+ print(remote_datasets)
Review Comment:
please remove `print` here
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] pankajkoti commented on pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "pankajkoti (via GitHub)" <gi...@apache.org>.
pankajkoti commented on PR #31360:
URL: https://github.com/apache/airflow/pull/31360#issuecomment-1588642007
cc: @utkarsharma2 could you please review this?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228400627
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
Review Comment:
Yes: https://github.com/apache/airflow/blob/9276310a43d17a9e9e38c2cb83686a15656896b2/airflow/providers/sftp/hooks/sftp.py#L95
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228179033
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
+ try:
+ remote_host = socket.gethostbyname(remote_host)
+ except Exception as e:
Review Comment:
Is it possible to handle more specific exceptions?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228410320
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
Review Comment:
Yes - but this also means that if OpenLineage is not installed, this should never get called. So the try-except here is superfluous, and, as in other PRs I've moved to checking this in call site.
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
+ try:
+ remote_host = socket.gethostbyname(remote_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve remote hostname. Using the provided hostname without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if hasattr(hook, "port"):
+ remote_port = hook.port
+ elif hasattr(hook, "ssh_hook"):
+ remote_port = hook.ssh_hook.port
+
+ # Since v4.1.0, SFTPOperator accepts both a string (single file) and a list of
+ # strings (multiple files) as local_filepath and remote_filepath, and internally
+ # keeps them as list in both cases. But before 4.1.0, only single string is
+ # allowed. So we consider both cases here for backward compatibility.
+ if isinstance(self.local_filepath, str):
+ local_filepath = [self.local_filepath]
+ else:
+ local_filepath = self.local_filepath
+ if isinstance(self.remote_filepath, str):
+ remote_filepath = [self.remote_filepath]
+ else:
+ remote_filepath = self.remote_filepath
+
+ local_datasets = [
+ Dataset(namespace=self._get_namespace(scheme, local_host, None, path), name=path)
+ for path in local_filepath
+ ]
+ remote_datasets = [
+ Dataset(namespace=self._get_namespace(scheme, remote_host, remote_port, path), name=path)
+ for path in remote_filepath
+ ]
+ print(remote_datasets)
Review Comment:
Done.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228174040
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
Review Comment:
based on your preference, not mandatory :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #31360:
URL: https://github.com/apache/airflow/pull/31360#issuecomment-1589661247
@phanikumv
> Can we add docs for this? how does a potential user know how to use the openlineage functionality along with sftp-operator?
As discussed with @sunank200 in other PR, I've added description on how OL data returned from here looks like.
As for user-facing per-operator docs, I don't see reason to or what would be in them - the point of OL integration is that it works automatically without user action on particular operator level, as long as user has "global" OL configuration set up - which in most cases is just the transport.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228164982
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
+ try:
+ remote_host = socket.gethostbyname(remote_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve remote hostname. Using the provided hostname without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if hasattr(hook, "port"):
+ remote_port = hook.port
+ elif hasattr(hook, "ssh_hook"):
+ remote_port = hook.ssh_hook.port
+
+ # Since v4.1.0, SFTPOperator accepts both a string (single file) and a list of
+ # strings (multiple files) as local_filepath and remote_filepath, and internally
+ # keeps them as list in both cases. But before 4.1.0, only single string is
+ # allowed. So we consider both cases here for backward compatibility.
+ if isinstance(self.local_filepath, str):
+ local_filepath = [self.local_filepath]
+ else:
+ local_filepath = self.local_filepath
+ if isinstance(self.remote_filepath, str):
+ remote_filepath = [self.remote_filepath]
+ else:
+ remote_filepath = self.remote_filepath
+
+ local_datasets = [
+ Dataset(namespace=self._get_namespace(scheme, local_host, None, path), name=path)
Review Comment:
I think this line will fail if the import failed `from openlineage.client.run import Dataset` right? Should we handle it better?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228176588
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
+ from openlineage.client.run import Dataset
+
+ from airflow.providers.openlineage.extractors import OperatorLineage
+ except ImportError:
+ return None
+
+ scheme = "file"
+ local_host = socket.gethostname()
+ try:
+ local_host = socket.gethostbyname(local_host)
+ except Exception as e:
+ self.log.warning(
+ f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}", # noqa: E501
+ exc_info=True,
+ )
+
+ if self.sftp_hook is not None:
+ hook = self.sftp_hook
+ elif self.ssh_hook is not None:
+ hook = self.ssh_hook
+ else:
+ hook = SFTPHook(ssh_conn_id=self.ssh_conn_id)
+
+ if self.remote_host is not None:
+ remote_host = self.remote_host
+ else:
+ remote_host = hook.get_connection(hook.ssh_conn_id).host
Review Comment:
If the hook is initialized from `self.sftp_hook` will it still have `hook.ssh_conn_id`??
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228191905
##########
tests/providers/sftp/operators/test_sftp.py:
##########
@@ -478,3 +481,112 @@ def test_return_str_when_local_filepath_was_str(self, mock_get):
return_value = sftp_op.execute(None)
assert isinstance(return_value, str)
assert return_value == local_filepath
+
Review Comment:
Should we also add a test case for missing imports case? ex - `from openlineage.client.run import Dataset`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski merged pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #31360:
URL: https://github.com/apache/airflow/pull/31360
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1272733605
##########
tests/providers/sftp/operators/test_sftp.py:
##########
@@ -478,3 +481,112 @@ def test_return_str_when_local_filepath_was_str(self, mock_get):
return_value = sftp_op.execute(None)
assert isinstance(return_value, str)
assert return_value == local_filepath
+
+ LOCAL_FILEPATH = "/path/local"
+ REMOTE_FILEPATH = "/path/remote"
+ LOCAL_DATASET = [
+ Dataset(namespace=f"file://{socket.gethostbyname(socket.gethostname())}:22", name=LOCAL_FILEPATH)
+ ]
+ REMOTE_DATASET = [Dataset(namespace="file://remotehost:22", name=REMOTE_FILEPATH)]
+
+ @pytest.mark.parametrize(
+ "operation, expected",
+ [
+ (SFTPOperation.GET, (REMOTE_DATASET, LOCAL_DATASET)),
+ (SFTPOperation.PUT, (LOCAL_DATASET, REMOTE_DATASET)),
+ ],
+ )
+ @mock.patch("airflow.providers.ssh.hooks.ssh.SSHHook.get_conn", spec=paramiko.SSHClient)
+ @mock.patch("airflow.providers.ssh.hooks.ssh.SSHHook.get_connection", spec=Connection)
+ def test_extract_ssh_conn_id(self, get_connection, get_conn, operation, expected):
+ get_connection.return_value = Connection(
+ conn_id="sftp_conn_id",
+ conn_type="sftp",
+ host="remotehost",
+ port=22,
+ )
+
+ dag_id = "sftp_dag"
+ task_id = "sftp_task"
+
+ task = SFTPOperator(
+ task_id=task_id,
+ ssh_conn_id="sftp_conn_id",
+ dag=DAG(dag_id),
+ start_date=timezone.utcnow(),
+ local_filepath="/path/local",
+ remote_filepath="/path/remote",
+ operation=operation,
+ )
+ lineage = task.get_openlineage_facets_on_start()
+
+ assert lineage.inputs == expected[0]
+ assert lineage.outputs == expected[1]
+
+ @pytest.mark.parametrize(
Review Comment:
Done.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228189249
##########
tests/providers/sftp/operators/test_sftp.py:
##########
@@ -478,3 +481,112 @@ def test_return_str_when_local_filepath_was_str(self, mock_get):
return_value = sftp_op.execute(None)
assert isinstance(return_value, str)
assert return_value == local_filepath
+
+ LOCAL_FILEPATH = "/path/local"
+ REMOTE_FILEPATH = "/path/remote"
+ LOCAL_DATASET = [
+ Dataset(namespace=f"file://{socket.gethostbyname(socket.gethostname())}:22", name=LOCAL_FILEPATH)
+ ]
+ REMOTE_DATASET = [Dataset(namespace="file://remotehost:22", name=REMOTE_FILEPATH)]
+
+ @pytest.mark.parametrize(
+ "operation, expected",
+ [
+ (SFTPOperation.GET, (REMOTE_DATASET, LOCAL_DATASET)),
+ (SFTPOperation.PUT, (LOCAL_DATASET, REMOTE_DATASET)),
+ ],
+ )
+ @mock.patch("airflow.providers.ssh.hooks.ssh.SSHHook.get_conn", spec=paramiko.SSHClient)
+ @mock.patch("airflow.providers.ssh.hooks.ssh.SSHHook.get_connection", spec=Connection)
+ def test_extract_ssh_conn_id(self, get_connection, get_conn, operation, expected):
+ get_connection.return_value = Connection(
+ conn_id="sftp_conn_id",
+ conn_type="sftp",
+ host="remotehost",
+ port=22,
+ )
+
+ dag_id = "sftp_dag"
+ task_id = "sftp_task"
+
+ task = SFTPOperator(
+ task_id=task_id,
+ ssh_conn_id="sftp_conn_id",
+ dag=DAG(dag_id),
+ start_date=timezone.utcnow(),
+ local_filepath="/path/local",
+ remote_filepath="/path/remote",
+ operation=operation,
+ )
+ lineage = task.get_openlineage_facets_on_start()
+
+ assert lineage.inputs == expected[0]
+ assert lineage.outputs == expected[1]
+
+ @pytest.mark.parametrize(
Review Comment:
```
[
(SFTPOperation.GET, (REMOTE_DATASET, LOCAL_DATASET)),
(SFTPOperation.PUT, (LOCAL_DATASET, REMOTE_DATASET)),
],
```
I see this repeating maybe we can refactor it?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] utkarsharma2 commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "utkarsharma2 (via GitHub)" <gi...@apache.org>.
utkarsharma2 commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228166941
##########
airflow/providers/sftp/operators/sftp.py:
##########
@@ -188,3 +191,85 @@ def execute(self, context: Any) -> str | list[str] | None:
raise AirflowException(f"Error while transferring {file_msg}, error: {str(e)}")
return self.local_filepath
+
+ def get_openlineage_facets_on_start(self):
+ try:
Review Comment:
just, curious is openlineage an optional package?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #31360: openlineage, sftp: add OpenLineage support for SFTPOperator
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #31360:
URL: https://github.com/apache/airflow/pull/31360#discussion_r1228414326
##########
tests/providers/sftp/operators/test_sftp.py:
##########
@@ -478,3 +481,112 @@ def test_return_str_when_local_filepath_was_str(self, mock_get):
return_value = sftp_op.execute(None)
assert isinstance(return_value, str)
assert return_value == local_filepath
+
Review Comment:
I think this should never happen - at least I do not see any reason or what would call those methods if OpenLineage is not installed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org