Programmer-RD-AI commented on code in PR #54259:
URL: https://github.com/apache/airflow/pull/54259#discussion_r2280571447
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py:
##########
@@ -235,20 +241,28 @@ class ElasticsearchPythonHook(BaseHook):
"""
Interacts with Elasticsearch. This hook uses the official Elasticsearch
Python Client.
+ .. deprecated:: 2.10.0
+ This hook is deprecated. Use
:class:`~airflow.providers.elasticsearch.hooks.elasticsearch.ElasticsearchHook`
instead.
+
:param hosts: list: A list of a single or many Elasticsearch instances.
Example: ["http://localhost:9200";]
:param es_conn_args: dict: Additional arguments you might need to enter to
connect to Elasticsearch.
Example: {"ca_cert":"/path/to/cert",
"basic_auth": "(user, pass)"}
"""
def __init__(self, hosts: list[Any], es_conn_args: dict | None = None):
Review Comment:
Optional: add `-> None` to make the return type explicit.
```suggestion
def __init__(self, hosts: list[Any], es_conn_args: dict | None = None)
-> None:
```
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py:
##########
@@ -268,3 +282,448 @@ def search(self, query: dict[Any, Any], index: str =
"_all") -> dict:
es_client = self.get_conn
result = es_client.search(index=index, body=query)
return result["hits"]
+
+
+class ElasticsearchHook(BaseHook):
+ """
+ Hook for interacting with Elasticsearch.
+
+ This hook provides a convenient interface to Elasticsearch operations
+ including search, bulk operations, and data management.
+ """
+
+ conn_name_attr = "elasticsearch_conn_id"
+ default_conn_name = "elasticsearch_default"
+ conn_type = "elasticsearch"
+ hook_name = "Elasticsearch"
+
+ def __init__(
+ self, elasticsearch_conn_id: str = "elasticsearch_default", log_query:
bool = False, **kwargs: Any
+ ) -> None:
+ """
+ Initialize the Elasticsearch Hook.
+
+ :param elasticsearch_conn_id: The Airflow connection ID for
Elasticsearch.
+ Default: *elasticsearch_default*.
+ :param log_query: If *True*, queries will be logged for debugging
purposes.
+ Default: *False*.
+ :param kwargs: Additional arguments passed to the parent BaseHook
class.
+ """
+ import os
+
+ super().__init__()
+ self.conn_id = elasticsearch_conn_id
+ self.log_query = log_query
+
+ # Environment variables as fallback configuration
+ self.vars: dict[str, str | bool | None] = {
+ "host": os.getenv("ELASTICSEARCH_HOST"),
+ "port": os.getenv("ELASTICSEARCH_PORT"),
+ "username": os.getenv("ELASTICSEARCH_USERNAME"),
+ "password": os.getenv("ELASTICSEARCH_PASSWORD"),
+ "use_ssl": os.getenv("ELASTICSEARCH_USE_SSL", "false").lower() ==
"true",
+ "verify_certs": os.getenv("ELASTICSEARCH_VERIFY_CERTS",
"true").lower() == "true",
+ "timeout": os.getenv("ELASTICSEARCH_TIMEOUT"),
+ "max_retries": os.getenv("ELASTICSEARCH_MAX_RETRIES"),
+ }
+
+ @cached_property
+ def conn(self) -> "AirflowConnection":
+ """Get the Airflow connection object for Elasticsearch."""
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def client(self) -> Elasticsearch:
+ """
+ Create and return an Elasticsearch client with environment variable
fallback.
+
+ Configuration priority:
+ 1. Airflow connection (highest priority)
+ 2. Environment variables (fallback)
+ 3. Default values (last resort)
+
+ :return: Configured Elasticsearch client instance.
+ :raises AirflowConfigException: If client creation fails due to
configuration issues.
+ """
+ # Get Airflow connection
+ conn = self.get_connection(self.conn_id)
+
+ # Configuration with fallback priority: connection -> env vars ->
defaults
+ host = conn.host or self.vars["host"] or "localhost"
+ port = conn.port or (int(self.vars["port"]) if self.vars["port"] else
None) or 9200
+ schema = conn.schema or ("https" if self.vars["use_ssl"] else "http")
+
+ # Build hosts list
+ hosts = [f"{schema}://{host}:{port}"]
+ client_args: dict[str, Any] = {"hosts": hosts}
Review Comment:
Optional: Using `Any` here makes the type a bit too loose.
Might be clearer to define a more specific type for `client_args` (e.g.
`dict[str, str | int | bool]` or a TypedDict)
so it’s easier to see what values are expected.
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py:
##########
@@ -268,3 +282,448 @@ def search(self, query: dict[Any, Any], index: str =
"_all") -> dict:
es_client = self.get_conn
result = es_client.search(index=index, body=query)
return result["hits"]
+
+
+class ElasticsearchHook(BaseHook):
+ """
+ Hook for interacting with Elasticsearch.
+
+ This hook provides a convenient interface to Elasticsearch operations
+ including search, bulk operations, and data management.
+ """
+
+ conn_name_attr = "elasticsearch_conn_id"
+ default_conn_name = "elasticsearch_default"
+ conn_type = "elasticsearch"
+ hook_name = "Elasticsearch"
+
+ def __init__(
+ self, elasticsearch_conn_id: str = "elasticsearch_default", log_query:
bool = False, **kwargs: Any
+ ) -> None:
+ """
+ Initialize the Elasticsearch Hook.
+
+ :param elasticsearch_conn_id: The Airflow connection ID for
Elasticsearch.
+ Default: *elasticsearch_default*.
+ :param log_query: If *True*, queries will be logged for debugging
purposes.
+ Default: *False*.
+ :param kwargs: Additional arguments passed to the parent BaseHook
class.
+ """
+ import os
+
+ super().__init__()
+ self.conn_id = elasticsearch_conn_id
+ self.log_query = log_query
+
+ # Environment variables as fallback configuration
+ self.vars: dict[str, str | bool | None] = {
+ "host": os.getenv("ELASTICSEARCH_HOST"),
+ "port": os.getenv("ELASTICSEARCH_PORT"),
+ "username": os.getenv("ELASTICSEARCH_USERNAME"),
+ "password": os.getenv("ELASTICSEARCH_PASSWORD"),
+ "use_ssl": os.getenv("ELASTICSEARCH_USE_SSL", "false").lower() ==
"true",
+ "verify_certs": os.getenv("ELASTICSEARCH_VERIFY_CERTS",
"true").lower() == "true",
+ "timeout": os.getenv("ELASTICSEARCH_TIMEOUT"),
+ "max_retries": os.getenv("ELASTICSEARCH_MAX_RETRIES"),
+ }
+
+ @cached_property
+ def conn(self) -> "AirflowConnection":
+ """Get the Airflow connection object for Elasticsearch."""
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def client(self) -> Elasticsearch:
+ """
+ Create and return an Elasticsearch client with environment variable
fallback.
+
+ Configuration priority:
+ 1. Airflow connection (highest priority)
+ 2. Environment variables (fallback)
+ 3. Default values (last resort)
+
+ :return: Configured Elasticsearch client instance.
+ :raises AirflowConfigException: If client creation fails due to
configuration issues.
+ """
+ # Get Airflow connection
+ conn = self.get_connection(self.conn_id)
+
+ # Configuration with fallback priority: connection -> env vars ->
defaults
+ host = conn.host or self.vars["host"] or "localhost"
+ port = conn.port or (int(self.vars["port"]) if self.vars["port"] else
None) or 9200
Review Comment:
Might be safer to use `.get()` when accessing `self.vars` (e.g.
`self.vars.get("host")`)
to avoid potential `KeyError`s if the key is missing. Keeps the same
fallback behavior
without assuming the key is always present.
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py:
##########
@@ -268,3 +282,448 @@ def search(self, query: dict[Any, Any], index: str =
"_all") -> dict:
es_client = self.get_conn
result = es_client.search(index=index, body=query)
return result["hits"]
+
+
+class ElasticsearchHook(BaseHook):
+ """
+ Hook for interacting with Elasticsearch.
+
+ This hook provides a convenient interface to Elasticsearch operations
+ including search, bulk operations, and data management.
+ """
+
+ conn_name_attr = "elasticsearch_conn_id"
+ default_conn_name = "elasticsearch_default"
+ conn_type = "elasticsearch"
+ hook_name = "Elasticsearch"
+
+ def __init__(
+ self, elasticsearch_conn_id: str = "elasticsearch_default", log_query:
bool = False, **kwargs: Any
Review Comment:
Optional: Instead of hardcoding `"elasticsearch_default"`, this could fall
back to the class-level
`default_conn_name`. That way the default only needs to be maintained in one
place.
```suggestion
self, elasticsearch_conn_id: str = None, log_query: bool = False,
**kwargs: Any
```
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py:
##########
@@ -268,3 +282,448 @@ def search(self, query: dict[Any, Any], index: str =
"_all") -> dict:
es_client = self.get_conn
result = es_client.search(index=index, body=query)
return result["hits"]
+
+
+class ElasticsearchHook(BaseHook):
+ """
+ Hook for interacting with Elasticsearch.
+
+ This hook provides a convenient interface to Elasticsearch operations
+ including search, bulk operations, and data management.
+ """
+
+ conn_name_attr = "elasticsearch_conn_id"
+ default_conn_name = "elasticsearch_default"
+ conn_type = "elasticsearch"
+ hook_name = "Elasticsearch"
+
+ def __init__(
+ self, elasticsearch_conn_id: str = "elasticsearch_default", log_query:
bool = False, **kwargs: Any
+ ) -> None:
+ """
+ Initialize the Elasticsearch Hook.
+
+ :param elasticsearch_conn_id: The Airflow connection ID for
Elasticsearch.
+ Default: *elasticsearch_default*.
+ :param log_query: If *True*, queries will be logged for debugging
purposes.
+ Default: *False*.
+ :param kwargs: Additional arguments passed to the parent BaseHook
class.
+ """
+ import os
+
+ super().__init__()
+ self.conn_id = elasticsearch_conn_id
+ self.log_query = log_query
+
+ # Environment variables as fallback configuration
+ self.vars: dict[str, str | bool | None] = {
+ "host": os.getenv("ELASTICSEARCH_HOST"),
+ "port": os.getenv("ELASTICSEARCH_PORT"),
+ "username": os.getenv("ELASTICSEARCH_USERNAME"),
+ "password": os.getenv("ELASTICSEARCH_PASSWORD"),
+ "use_ssl": os.getenv("ELASTICSEARCH_USE_SSL", "false").lower() ==
"true",
+ "verify_certs": os.getenv("ELASTICSEARCH_VERIFY_CERTS",
"true").lower() == "true",
+ "timeout": os.getenv("ELASTICSEARCH_TIMEOUT"),
+ "max_retries": os.getenv("ELASTICSEARCH_MAX_RETRIES"),
+ }
+
+ @cached_property
+ def conn(self) -> "AirflowConnection":
+ """Get the Airflow connection object for Elasticsearch."""
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def client(self) -> Elasticsearch:
+ """
+ Create and return an Elasticsearch client with environment variable
fallback.
+
+ Configuration priority:
+ 1. Airflow connection (highest priority)
+ 2. Environment variables (fallback)
+ 3. Default values (last resort)
+
+ :return: Configured Elasticsearch client instance.
+ :raises AirflowConfigException: If client creation fails due to
configuration issues.
+ """
+ # Get Airflow connection
+ conn = self.get_connection(self.conn_id)
+
+ # Configuration with fallback priority: connection -> env vars ->
defaults
+ host = conn.host or self.vars["host"] or "localhost"
+ port = conn.port or (int(self.vars["port"]) if self.vars["port"] else
None) or 9200
+ schema = conn.schema or ("https" if self.vars["use_ssl"] else "http")
Review Comment:
For consistency with `host` and `port`, the `schema` fallback could follow
the same
`conn → vars → default` structure, e.g.:
```suggestion
schema = conn.schema or ("https" if self.vars.get("use_ssl") else
None) or "http"
```
--
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]
