ashb commented on code in PR #30735:
URL: https://github.com/apache/airflow/pull/30735#discussion_r1171939263
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -70,9 +74,34 @@ def __init__(self, client: OpenLineageClient | None = None,
secrets_masker: Secr
def get_or_create_openlineage_client(self) -> OpenLineageClient:
if not self._client:
- self._client = OpenLineageClient.from_environment()
+ config = self.get_openlineage_config()
+ if config:
+ self._client = OpenLineageClient.from_dict(config=config)
+ else:
+ self._client = OpenLineageClient.from_environment()
return self._client
+ def get_openlineage_config(self) -> dict | None:
+ # First, try to read from YAML file
+ openlineage_config_path = conf.get("openlineage", "config_path")
+ if openlineage_config_path:
+ config = self._read_yaml_config(openlineage_config_path)
+ if config:
+ return config.get("transport", None)
+ # Second, try to get transport config
+ transport = conf.getjson("openlineage", "transport")
+ if not transport:
+ return None
+ elif not isinstance(transport, dict):
+ raise ValueError(f"{transport} is not a dict")
+ return transport
+
+ def _read_yaml_config(self, path: str) -> dict | None:
+ if os.path.exists(path):
Review Comment:
If a something is specified but doesn't exist it should probably be a error?
```suggestion
if path:
```
--
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]
