kacpermuda commented on code in PR #74805:
URL: https://github.com/apache/airflow/pull/74805#discussion_r4380792510
##########
providers/common/sql/src/airflow/providers/common/sql/operators/sql.py:
##########
@@ -337,11 +343,16 @@ def get_openlineage_facets_on_complete(self,
task_instance) -> OperatorLineage |
operator_lineage = self.get_openlineage_facets_on_start() or
OperatorLineage()
Review Comment:
Fixed. `on_start` now builds the engine inside `try/except Exception` (debug
log, falls back to `None`), with a test. I kept it ungated rather than only
when `use_external_connection`, since the engine is also used to compile SQL. I
didn't switch to the compat function here, to keep the PR scoped.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @kacpermuda before posting
##########
providers/openlineage/src/airflow/providers/openlineage/sqlparser.py:
##########
@@ -489,23 +489,29 @@ def get_openlineage_facets_with_sql(
connection = hook.get_connection(conn_id)
if database_info is None:
- try:
- database_info = hook.get_openlineage_database_info(connection)
- except AttributeError:
- database_info = None
+ get_database_info = getattr(hook, "get_openlineage_database_info",
None)
+ if get_database_info is not None:
+ try:
+ database_info = get_database_info(connection)
+ except Exception as e:
+ log.warning("Failed to get OpenLineage database info for %s:
%s", hook, e)
+ log.debug("OpenLineage failure details:", exc_info=True)
if database_info is None:
log.debug("%s has no database info provided", hook)
return None
- try:
- sql_parser = SQLParser(
- dialect=hook.get_openlineage_database_dialect(connection),
- default_schema=hook.get_openlineage_default_schema(),
- )
- except AttributeError:
+ get_dialect = getattr(hook, "get_openlineage_database_dialect", None)
+ get_default_schema = getattr(hook, "get_openlineage_default_schema", None)
+ if get_dialect is None or get_default_schema is None:
log.debug("%s failed to get database dialect", hook)
return None
+ try:
+ sql_parser = SQLParser(dialect=get_dialect(connection),
default_schema=get_default_schema())
+ except Exception as e:
+ log.warning("Failed to get OpenLineage database dialect for %s: %s",
hook, e)
Review Comment:
Done. The message now says "database dialect or default schema" and uses the
conn id, in all copies (operator, `sqlparser`, compat fallback).
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @kacpermuda before posting
##########
providers/common/sql/tests/unit/common/sql/operators/test_sql_execute.py:
##########
@@ -368,6 +368,70 @@ def get_db_hook(self):
)
+def test_openlineage_hook_without_database_info_method_returns_empty_lineage():
Review Comment:
Added `assert_not_called()` on the warning and switched to the decorator
form.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @kacpermuda before posting
##########
providers/common/compat/tests/unit/common/compat/openlineage/utils/test_sql.py:
##########
@@ -16,8 +16,67 @@
# under the License.
from __future__ import annotations
+import importlib
+import sys
+import types
+from unittest import mock
+
+import pytest
+
+_MODULE = "airflow.providers.common.compat.openlineage.utils.sql"
+
def test_import():
from airflow.providers.common.compat.openlineage.utils.sql import
get_openlineage_facets_with_sql
assert get_openlineage_facets_with_sql is not None
+
+
[email protected]
+def fallback_get_openlineage_facets_with_sql():
+ """Load the module as if the OpenLineage provider was too old to ship
``get_openlineage_facets_with_sql``."""
+ old_sqlparser = types.ModuleType("airflow.providers.openlineage.sqlparser")
+ old_sqlparser.SQLParser = mock.MagicMock(name="SQLParser")
+ with mock.patch.dict(sys.modules,
{"airflow.providers.openlineage.sqlparser": old_sqlparser}):
Review Comment:
Switched to `monkeypatch.delattr`. Ran this file followed by
`test_utils.py`, which passes.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @kacpermuda before posting
--
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]
