You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "cjames23 (via GitHub)" <gi...@apache.org> on 2023/10/02 05:02:42 UTC
[GitHub] [airflow] cjames23 opened a new pull request, #34705: Open Search Provider
cjames23 opened a new pull request, #34705:
URL: https://github.com/apache/airflow/pull/34705
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
This is a Draft PR to get feedback before putting out a lazy consensus to the dev list for a new provider for Open Search. This provider would be a pre-requisite for adding functionality to the Amazon provider for the AWS managed service for OpenSearch. This was discussed in a different [PR](https://github.com/apache/airflow/pull/34693) which will be blocked by this PR now. both PRs would be pre-work for completing https://github.com/apache/airflow/issues/33619 and adding Open Search log integration.
### Open questions to more experienced contributors:
1. For system test DAG to succeed would an OpenSearch domain need to be setup and if so what is the approach for standing up such infrastructure for integration tests?
2. Any advice on mypy for some of the operator parameters? The goal is to allow users to either pass in the arguments directly or by passing in a class in the case of Search which works but in the case of adding a document the end user of the Operator needs to subclass the Document class. I am not sure what the best approach is for the typing for that parameter.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1745014377
> For system test DAG to succeed would an OpenSearch domain need to be setup and if so what is the approach for standing up such infrastructure for integration tests?
System tests are not triggered as part of the CI. They are just used as example and also as system test/integration test if anybody wants to execute it (and as such, needs to setup the infra). The approach taken by the community is your write system tests in the Airflow repository but you execute them on your own with your infra set up. As an example, in the Amazon provider package, all system tests rely on AWS services (thus AWS account). These system tests are not triggered as part of Airflow CI but AWS team trigger frequently and results are accessible through a [dashboard](https://aws-mwaa.github.io/open-source/system-tests/dashboard.html).
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1752191814
Lazy consensus has been reached on adding this new provider from this PR.
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1760864875
> Yeah, I saw that, but the only changes in the Amazon package were some whitespace and punctuation in a docstring. Better done separate, but I didn't think it was worth blocking over it.
Those changes were because pre-commit checks failed locally for those things after a merge pull.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1344300110
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
Hmm but it is still failing that check. Okay I’ll move that back to 2.5.0
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
Hmm but it is still failing that check. Okay I’ll move that back to 2.5.0
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1763061002
> If the checks fail after this merge from main do you want me to include any fixes here for the breeze mypy issue? I think it just needs to be changed from ignore[misc] to ignore[attr-defined]. Not sure about the unit tests that are failing though from core.
We have identified the issue
https://github.com/apache/airflow/pull/34941 should fix 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
Re: [PR] Add Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1760851926
@eladkal any advice on the failure that is happening with the CI image build step and the static checks? I ran breeze static checks against all files locally and do not receive the same failure.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1344298941
##########
generated/provider_dependencies.json:
##########
@@ -28,6 +28,10 @@
"boto3>=1.28.0",
"botocore>=1.31.0",
"jsonpath_ng>=1.5.3",
+ "mypy-boto3-appflow>=1.24.0",
+ "mypy-boto3-rds>=1.24.0",
+ "mypy-boto3-redshift-data>=1.24.0",
+ "mypy-boto3-s3>=1.24.0",
Review Comment:
Good question. Let me check through things the only dependency defined that I removed was OpenSearch-dsl because after 2.1.0 of OpenSearch-py that was added in directly to have a single 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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347711520
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
Review Comment:
Nitpick, and doesn't REALLY matter since it's a unit test, but this would parse out to "unit_teststest_schedule_dag_once", note the second "word" gets smooshed.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347702655
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
Does this actually work if you only run the tests in this file? I wold have expected to need to import the mock_hook fixture
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1357592192
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ Provide a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def conn(self):
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
Review Comment:
I had to push changes for breeze commands so I am pushing this change now.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1348704834
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ Provide a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def conn(self):
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
Review Comment:
nit: since it is a property, I would just call it `client`
```suggestion
def client(self) -> OpenSearch:
```
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1745954349
@vincbeck Thank you for the clarification on the example dags in the systems tests.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347706512
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
+ result = OpenSearchHook(open_search_conn_id="opensearch_default", log_query=True).search(
+ index_name="testIndex",
+ query={"size": 1, "query": {"multi_match": {"query": "test", "fields": ["testField"]}}},
+ )
Review Comment:
[Here and elsewhere] Functionally identical but perhaps a little easier to read if you split the hook out into its own assignment? Not sure, I'll leave it up to you
```suggestion
hook = OpenSearchHook(open_search_conn_id="opensearch_default", log_query=True)
result = hook.search(
index_name="testIndex",
query={"size": 1, "query": {"multi_match": {"query": "test", "fields": ["testField"]}}},
)
```
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1752744510
Just to clarify - This PR suppose to be only open search provider related code and https://github.com/apache/airflow/pull/34693 should be a followup with additional integrations for open search with other providers right?
Asking because if so then this PR should not modify Amazon provider code
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347690886
##########
tests/providers/opensearch/conftest.py:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from airflow.models import Connection
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+from airflow.utils import db
+
+MOCK_RETURN = {"status": "test"}
+
+
+class MockSearch(OpenSearchHook):
+ # Mock class to override the Hook for monkeypatching
+ def get_client(self) -> None:
+ return None
+
+ def search(self, query: dict, index_name: str, **kwargs: Any) -> Any:
+ return MOCK_RETURN
+
+ def index(self, document: dict, index_name: str, doc_id: int, **kwargs: Any) -> Any:
+ return doc_id
+
+
+@pytest.fixture
+def mock_hook(monkeypatch):
+ monkeypatch.setattr(OpenSearchHook, "search", MockSearch.search)
+ monkeypatch.setattr(OpenSearchHook, "get_client", MockSearch.get_client)
+ monkeypatch.setattr(OpenSearchHook, "index", MockSearch.index)
+
+
+@pytest.fixture(autouse=True)
+def setup_connection():
+ # We need to set up a Connection into the database for all tests.
+ db.merge_conn(
+ Connection(
+ conn_id="opensearch_default",
+ conn_type="opensearch",
+ host="myopensearch.com",
+ login="test_usser",
Review Comment:
I know it doesn't actually matter, but typo in `usser`
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346603646
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ This Hook provides a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.conn = self.get_connection(self.conn_id)
Review Comment:
Good call out. Pushed change for it to be a cached_property
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1755950688
> Yeah, I saw that, but the only changes in the Amazon package were some whitespace and punctuation in a docstring. Better done separate, but I didn't think it was worth blocking over it.
Not a blocker but since rebase and resolving conflicts is required it can be done along with 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
Re: [PR] Add Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1762208402
What an odd static check to fail, I can't see how it could be related. I'm going to bump it and see if it was just flake. If it fails again, we'll have to sort it out.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1750034991
> @ferruzzi those should not have any impact. I think they showed up because I didn’t pay attention when creating this new branch as my local was still on the chime notifier branch. I can try to cherry pick them out if there is a blocking concern but all of that code was already merged.
I think would be easier to just squash all commits
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1344387696
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
That is OK. This PR is depended on https://github.com/apache/airflow/pull/34728
Just set 2.5.0 and when we merge https://github.com/apache/airflow/pull/34728 it will fix the issue
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1765090521
Nice, congrats!
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "pateash (via GitHub)" <gi...@apache.org>.
pateash commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1778986522
Hi @cjames23,
Just a quick question regarding this one.
as we know OpenSearch is just a fork of Elasticsearch which happened couple of years ago,
Can we not use Elastic search operators for connecting to OpenSearch,
and if these hooks and operators are very similar to elastic search operators.
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1763467597
🎉 🎉 🎉 🎉 🎉 🎉
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal merged PR #34705:
URL: https://github.com/apache/airflow/pull/34705
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1749765268
I just noticed in the commits tab there are a couple of "chime notifier" commit messages. Those should likely not be in the PR?
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346599556
##########
airflow/providers/opensearch/operators/opensearch.py:
##########
@@ -0,0 +1,192 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class OpenSearchQueryOperator(BaseOperator):
+ """
+ Runs a query search against a given index on an OpenSearch cluster and returns results.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchQueryOperator`
+
+ :param: query: A Dictionary Open Search DSL query.
+ :param: search_object: A Search object from opensearch-dsl.
+ :param: index_name: The name of the index to search for documents.
+ :param: opensearch_conn_id: opensearch connection to use
+ :param: log_query: Whether to log the query used. Defaults to True and logs query used.
+ """
+
+ template_fields: Sequence[str] = ["query"]
+
+ def __init__(
+ self,
+ *,
+ query: dict | None = None,
+ search_object: Any | None = None,
+ index_name: str | None = None,
+ opensearch_conn_id: str = "opensearch_default",
+ log_query: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.query = query
+ self.index_name = index_name
+ self.opensearch_conn_id = opensearch_conn_id
+ self.log_query = log_query
+ self.search_object = search_object
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=self.log_query)
+
+ def execute(self, context: Context) -> Any:
+ """Executes a search against a given index or a Search object on an OpenSearch Cluster."""
+ result = None
+
+ if self.query is not None:
+ if not self.query.get("query"):
+ raise AirflowException("Query input is missing required field Query in dictionary")
+ if self.index_name is None:
+ raise AirflowException("Index name is required when using the query input.")
+ try:
+ result = self.hook.search(index_name=self.index_name, query=self.query)
+ except Exception as e:
+ raise AirflowException(e)
+ elif self.search_object is not None:
+ try:
+ result = self.search_object.using(self.hook.get_client).execute()
+ except Exception as e:
+ raise AirflowException(e)
+ else:
+ raise AirflowException(
+ """Input missing required input of query or search_object.
+ Either query or search_object is required."""
+ )
+ return result
+
+
+class OpenSearchCreateIndexOperator(BaseOperator):
+ """
+ Creates a new index on an Open Search cluster with a given index name.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchCreateIndexOperator`
+
+ :param: index_name: The name of the index to be created.
+ :param: index_body: A dictionary that defines index settings
+ :param: opensearch_conn_id: opensearch connection to use
+ """
+
+ def __init__(
+ self,
+ *,
+ index_name: str,
+ index_body: dict[str, Any],
+ opensearch_conn_id: str = "opensearch_default",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.index_name = index_name
+ self.index_body = index_body
+ self.opensearch_conn_id = opensearch_conn_id
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=False)
+
+ def execute(self, context: Context) -> Any:
+ """Creates an index on an Open Search cluster."""
+ try:
+ self.hook.get_client.indices.create(index=self.index_name, body=self.index_body)
+ except Exception as e:
+ raise AirflowException(e)
+
+
+class OpenSearchAddDocumentOperator(BaseOperator):
+ """
+ Adds a new document to a given Index. It will either add or overwrite an existing document.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchAddDocumentOperator`
+
+ :param: index_name: The name of the index to put the document.
+ :param: document: A dictionary representation of the document.
+ :param: document_id: The id for the document in the index.
+ :param: doc_class: A Document subclassed object using opensearch-dsl
+ :param: opensearch_conn_id: opensearch connection to use
+ """
+
+ def __init__(
+ self,
+ *,
+ index_name: str | None = None,
+ document: dict[str, Any] | None = None,
+ doc_id: int | None = None,
+ doc_class: Any | None = None,
+ opensearch_conn_id: str = "opensearch_default",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.index_name = index_name
+ self.document = document
+ self.doc_id = doc_id
+ self.doc_class = doc_class
+ self.opensearch_conn_id = opensearch_conn_id
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=False)
+
+ def execute(self, context: Context) -> Any:
+ """Saves a document to a given index on an OpenSearch cluster."""
+ if self.doc_class is not None:
+ try:
+ doc = self.doc_class.init(using=self.hook.get_client)
+ result = doc.save(using=self.hook.get_client)
+ except Exception as e:
Review Comment:
Fixed to use the base OpenSearchException for the OpenSearchClient instead. It would get very verbose otherwise to go to individual exceptions as there are a potential of 15 different exceptions possible.
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1755924053
Yeah, I saw that, but the only changes in the Amazon package were some whitespace and punctuation in a docstring. Better done separate, but I didn't think it was worth blocking over 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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347691774
##########
tests/providers/opensearch/conftest.py:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from airflow.models import Connection
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+from airflow.utils import db
+
+MOCK_RETURN = {"status": "test"}
+
+
Review Comment:
I haven't seen monkeypatch used this way, I really like that.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1344387696
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
That is OK. This PR is depended on https://github.com/apache/airflow/pull/34728
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347714300
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
+
+ self.open_search = OpenSearchQueryOperator(
+ task_id="test_opensearch_query_operator",
+ index_name="test_index",
+ query={
+ "size": 5,
+ "query": {"multi_match": {"query": "test", "fields": ["test_title^2", "test_type"]}},
+ },
+ )
+
+ def test_init(self):
+ assert self.open_search.task_id == "test_opensearch_query_operator"
+ assert self.open_search.opensearch_conn_id == "opensearch_default"
+ assert self.open_search.query["size"] == 5
+
+ def test_search_query(self, mock_hook):
+ result = self.open_search.execute({})
+ assert result == MOCK_SEARCH_RETURN
Review Comment:
Consider renaming MOCK_SEARCH_RETURN to EXPECTED_SEARCH_RETURN perhaps?
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347751754
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
+
+ self.open_search = OpenSearchQueryOperator(
+ task_id="test_opensearch_query_operator",
+ index_name="test_index",
+ query={
+ "size": 5,
+ "query": {"multi_match": {"query": "test", "fields": ["test_title^2", "test_type"]}},
+ },
+ )
+
+ def test_init(self):
+ assert self.open_search.task_id == "test_opensearch_query_operator"
+ assert self.open_search.opensearch_conn_id == "opensearch_default"
+ assert self.open_search.query["size"] == 5
+
+ def test_search_query(self, mock_hook):
+ result = self.open_search.execute({})
+ assert result == MOCK_SEARCH_RETURN
+
+
+class TestOpenSearchCreateIndexOperator:
+ # This test does not test execute logic because there is only a redirect to the OpenSearch
+ # client.
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
Review Comment:
Good callout. Moved this to a fixture in the latest push.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1344279037
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
If I do that then it fails the CI check for Airflow 2.4
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1763123380
Fixed https://github.com/apache/airflow/pull/34941 and few other issues that caused broken main. Rebasing/resolving conflicts (generally regenerating breeze images with. `breeze setup regenerate-images` should fix 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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347734216
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
Here is a output in breeze running just a single test.
```
root@a8147a85fe4a:/opt/airflow/tests/providers/opensearch/hooks# pytest test_opensearch.py -k "test_hook_search"
========================================= test session starts =========================================
platform linux -- Python 3.10.13, pytest-7.4.2, pluggy-1.3.0 -- /usr/local/bin/python
cachedir: .pytest_cache
rootdir: /opt/airflow
configfile: pyproject.toml
plugins: httpx-0.21.3, mock-3.11.1, xdist-3.3.1, capture-warnings-0.0.4, instafail-0.5.0, anyio-4.0.0, rerunfailures-12.0, timeouts-1.2.1, asyncio-0.21.1, cov-4.1.0, time-machine-2.13.0, requests-mock-1.11.0
asyncio: mode=strict
setup timeout: 0.0s, execution timeout: 0.0s, teardown timeout: 0.0s
collected 2 items / 1 deselected / 1 selected
test_opensearch.py::TestOpenSearchHook::test_hook_search PASSED [100%]
=================================== 1 passed, 1 deselected in 0.43s ===================================
```
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347724579
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
If it works, it works. I must have my scopes mixed up. My mistake.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347750625
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
Review Comment:
Fixed in latest push.
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
+
+ self.open_search = OpenSearchQueryOperator(
+ task_id="test_opensearch_query_operator",
+ index_name="test_index",
+ query={
+ "size": 5,
+ "query": {"multi_match": {"query": "test", "fields": ["test_title^2", "test_type"]}},
+ },
+ )
+
+ def test_init(self):
+ assert self.open_search.task_id == "test_opensearch_query_operator"
+ assert self.open_search.opensearch_conn_id == "opensearch_default"
+ assert self.open_search.query["size"] == 5
+
+ def test_search_query(self, mock_hook):
+ result = self.open_search.execute({})
+ assert result == MOCK_SEARCH_RETURN
Review Comment:
Renamed in latest push.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347750353
##########
tests/providers/opensearch/conftest.py:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from airflow.models import Connection
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+from airflow.utils import db
+
+MOCK_RETURN = {"status": "test"}
+
+
+class MockSearch(OpenSearchHook):
+ # Mock class to override the Hook for monkeypatching
+ def get_client(self) -> None:
+ return None
+
+ def search(self, query: dict, index_name: str, **kwargs: Any) -> Any:
+ return MOCK_RETURN
+
+ def index(self, document: dict, index_name: str, doc_id: int, **kwargs: Any) -> Any:
+ return doc_id
+
+
+@pytest.fixture
+def mock_hook(monkeypatch):
+ monkeypatch.setattr(OpenSearchHook, "search", MockSearch.search)
+ monkeypatch.setattr(OpenSearchHook, "get_client", MockSearch.get_client)
+ monkeypatch.setattr(OpenSearchHook, "index", MockSearch.index)
+
+
+@pytest.fixture(autouse=True)
+def setup_connection():
+ # We need to set up a Connection into the database for all tests.
+ db.merge_conn(
+ Connection(
+ conn_id="opensearch_default",
+ conn_type="opensearch",
+ host="myopensearch.com",
+ login="test_usser",
Review Comment:
Fixed in latest push.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346588765
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ This Hook provides a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.conn = self.get_connection(self.conn_id)
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
+ """This function is intended for Operators that forward high level client objects."""
+ auth = (self.conn.login, self.conn.password)
+ client = OpenSearch(
+ hosts=[{"host": self.conn.host, "port": self.conn.port}],
+ http_auth=auth,
+ use_ssl=self.use_ssl,
+ verify_certs=self.verify_certs,
+ connection_class=RequestsHttpConnection,
+ )
+ return client
+
+ def search(self, query: dict, index_name: str, **kwargs: Any) -> Any:
+ """
+ Runs a search query against the connected OpenSearch cluster.
+
+ :param: query: The query for the search against OpenSearch.
+ :param: index_name: The name of the index to search against
+ """
+ if self.log_query:
+ self.log.info("Searching %s with Query: %s", index_name, query)
+ return self.get_client.search(body=query, index=index_name, **kwargs)
+
+ def index(self, document: dict, index_name: str, doc_id: int, **kwargs: Any) -> Any:
+ """
+ Index a document on open search.
Review Comment:
No it should be OpenSearch I missed changing this back once I added opensearch and OpenSearch to the spelling list to avoid spelling linting errors.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346600199
##########
docs/apache-airflow-providers-opensearch/connections/opensearch.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+
+OpenSearch Connection
+=====================
+The Open Search connection provides credentials for an OpenSearch instance.
+
+Configuring the Connection
+--------------------------
+Host (required)
+ The host address of the Open Search instance.
+Login (required)
+ The login user.
+Password (required)
+ The password for the login user.
+Extra (optional)
+ Specifying the extra parameters as a (json dictionary) that can be used in the Open Search connection.
+ The following parameters are all optional:
+
+ * ``use_ssl``: Boolean on requiring an ssl connection.
+ * ``verify_certs``: Boolean indicating to verify certs for ssl.
Review Comment:
Added what the default is set to.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346323410
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ This Hook provides a thin wrapper around the OpenSearch client.
Review Comment:
[Here and elsewhere] This is a very minor non-blocking nitpick, you can ignore if you wish. One of my side projects I've been chipping away at is enabling more pydocstyle checks to help standardize our codebase. One of the ones I am working on enabling enforces the rule that the first line in every docstring needs to be phrased as an imperative. For example:
```suggestion
Provide a thin wrapper around the OpenSearch client.
```
Basically, phrase it as if you are telling the method or class what to do, not explaining it to someone else. If you could make the adjustments, I'd appreciate it. If not, I can go back and do them when I get to it. I won't block merging for this.
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ This Hook provides a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.conn = self.get_connection(self.conn_id)
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
+ """This function is intended for Operators that forward high level client objects."""
+ auth = (self.conn.login, self.conn.password)
+ client = OpenSearch(
+ hosts=[{"host": self.conn.host, "port": self.conn.port}],
+ http_auth=auth,
+ use_ssl=self.use_ssl,
+ verify_certs=self.verify_certs,
+ connection_class=RequestsHttpConnection,
+ )
+ return client
+
+ def search(self, query: dict, index_name: str, **kwargs: Any) -> Any:
+ """
+ Runs a search query against the connected OpenSearch cluster.
+
+ :param: query: The query for the search against OpenSearch.
+ :param: index_name: The name of the index to search against
+ """
+ if self.log_query:
+ self.log.info("Searching %s with Query: %s", index_name, query)
+ return self.get_client.search(body=query, index=index_name, **kwargs)
+
+ def index(self, document: dict, index_name: str, doc_id: int, **kwargs: Any) -> Any:
+ """
+ Index a document on open search.
Review Comment:
Is this correct or should it be `OpenSearch`?
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ This Hook provides a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.conn = self.get_connection(self.conn_id)
Review Comment:
Perhaps consider making conn a cached_property? I know the Amazon provider caches it, I don't know if it's appropriate here or not. I'll leave it up to you.
##########
airflow/providers/opensearch/operators/opensearch.py:
##########
@@ -0,0 +1,192 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class OpenSearchQueryOperator(BaseOperator):
+ """
+ Runs a query search against a given index on an OpenSearch cluster and returns results.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchQueryOperator`
+
+ :param: query: A Dictionary Open Search DSL query.
+ :param: search_object: A Search object from opensearch-dsl.
+ :param: index_name: The name of the index to search for documents.
+ :param: opensearch_conn_id: opensearch connection to use
+ :param: log_query: Whether to log the query used. Defaults to True and logs query used.
+ """
+
+ template_fields: Sequence[str] = ["query"]
+
+ def __init__(
+ self,
+ *,
+ query: dict | None = None,
+ search_object: Any | None = None,
+ index_name: str | None = None,
+ opensearch_conn_id: str = "opensearch_default",
+ log_query: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.query = query
+ self.index_name = index_name
+ self.opensearch_conn_id = opensearch_conn_id
+ self.log_query = log_query
+ self.search_object = search_object
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=self.log_query)
+
+ def execute(self, context: Context) -> Any:
+ """Executes a search against a given index or a Search object on an OpenSearch Cluster."""
+ result = None
+
+ if self.query is not None:
+ if not self.query.get("query"):
+ raise AirflowException("Query input is missing required field Query in dictionary")
+ if self.index_name is None:
+ raise AirflowException("Index name is required when using the query input.")
+ try:
+ result = self.hook.search(index_name=self.index_name, query=self.query)
+ except Exception as e:
+ raise AirflowException(e)
+ elif self.search_object is not None:
+ try:
+ result = self.search_object.using(self.hook.get_client).execute()
+ except Exception as e:
+ raise AirflowException(e)
+ else:
+ raise AirflowException(
+ """Input missing required input of query or search_object.
+ Either query or search_object is required."""
+ )
+ return result
+
+
+class OpenSearchCreateIndexOperator(BaseOperator):
+ """
+ Creates a new index on an Open Search cluster with a given index name.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchCreateIndexOperator`
+
+ :param: index_name: The name of the index to be created.
+ :param: index_body: A dictionary that defines index settings
+ :param: opensearch_conn_id: opensearch connection to use
+ """
+
+ def __init__(
+ self,
+ *,
+ index_name: str,
+ index_body: dict[str, Any],
+ opensearch_conn_id: str = "opensearch_default",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.index_name = index_name
+ self.index_body = index_body
+ self.opensearch_conn_id = opensearch_conn_id
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=False)
+
+ def execute(self, context: Context) -> Any:
+ """Creates an index on an Open Search cluster."""
+ try:
+ self.hook.get_client.indices.create(index=self.index_name, body=self.index_body)
+ except Exception as e:
+ raise AirflowException(e)
+
+
+class OpenSearchAddDocumentOperator(BaseOperator):
+ """
+ Adds a new document to a given Index. It will either add or overwrite an existing document.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchAddDocumentOperator`
+
+ :param: index_name: The name of the index to put the document.
+ :param: document: A dictionary representation of the document.
+ :param: document_id: The id for the document in the index.
+ :param: doc_class: A Document subclassed object using opensearch-dsl
+ :param: opensearch_conn_id: opensearch connection to use
+ """
+
+ def __init__(
+ self,
+ *,
+ index_name: str | None = None,
+ document: dict[str, Any] | None = None,
+ doc_id: int | None = None,
+ doc_class: Any | None = None,
+ opensearch_conn_id: str = "opensearch_default",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.index_name = index_name
+ self.document = document
+ self.doc_id = doc_id
+ self.doc_class = doc_class
+ self.opensearch_conn_id = opensearch_conn_id
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=False)
+
+ def execute(self, context: Context) -> Any:
+ """Saves a document to a given index on an OpenSearch cluster."""
+ if self.doc_class is not None:
+ try:
+ doc = self.doc_class.init(using=self.hook.get_client)
+ result = doc.save(using=self.hook.get_client)
+ except Exception as e:
Review Comment:
[Here and below] I'd prefer to see a more narrow catch here. Catching `Exception` is generally considered poor form, in my experience.
##########
docs/apache-airflow-providers-opensearch/connections/opensearch.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+
+OpenSearch Connection
+=====================
+The Open Search connection provides credentials for an OpenSearch instance.
+
+Configuring the Connection
+--------------------------
+Host (required)
+ The host address of the Open Search instance.
+Login (required)
+ The login user.
+Password (required)
+ The password for the login user.
+Extra (optional)
+ Specifying the extra parameters as a (json dictionary) that can be used in the Open Search connection.
+ The following parameters are all optional:
+
+ * ``use_ssl``: Boolean on requiring an ssl connection.
+ * ``verify_certs``: Boolean indicating to verify certs for ssl.
Review Comment:
[Optional] Consider including the default values in the docs here.
##########
airflow/providers/opensearch/operators/opensearch.py:
##########
@@ -0,0 +1,192 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class OpenSearchQueryOperator(BaseOperator):
+ """
+ Runs a query search against a given index on an OpenSearch cluster and returns results.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchQueryOperator`
+
+ :param: query: A Dictionary Open Search DSL query.
+ :param: search_object: A Search object from opensearch-dsl.
+ :param: index_name: The name of the index to search for documents.
+ :param: opensearch_conn_id: opensearch connection to use
+ :param: log_query: Whether to log the query used. Defaults to True and logs query used.
+ """
+
+ template_fields: Sequence[str] = ["query"]
+
+ def __init__(
+ self,
+ *,
+ query: dict | None = None,
+ search_object: Any | None = None,
+ index_name: str | None = None,
+ opensearch_conn_id: str = "opensearch_default",
+ log_query: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.query = query
+ self.index_name = index_name
+ self.opensearch_conn_id = opensearch_conn_id
+ self.log_query = log_query
+ self.search_object = search_object
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=self.log_query)
+
+ def execute(self, context: Context) -> Any:
+ """Executes a search against a given index or a Search object on an OpenSearch Cluster."""
+ result = None
+
+ if self.query is not None:
+ if not self.query.get("query"):
+ raise AirflowException("Query input is missing required field Query in dictionary")
+ if self.index_name is None:
+ raise AirflowException("Index name is required when using the query input.")
+ try:
+ result = self.hook.search(index_name=self.index_name, query=self.query)
+ except Exception as e:
+ raise AirflowException(e)
+ elif self.search_object is not None:
+ try:
+ result = self.search_object.using(self.hook.get_client).execute()
+ except Exception as e:
+ raise AirflowException(e)
+ else:
+ raise AirflowException(
+ """Input missing required input of query or search_object.
+ Either query or search_object is required."""
+ )
+ return result
+
+
+class OpenSearchCreateIndexOperator(BaseOperator):
+ """
+ Creates a new index on an Open Search cluster with a given index name.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:OpenSearchCreateIndexOperator`
+
+ :param: index_name: The name of the index to be created.
+ :param: index_body: A dictionary that defines index settings
+ :param: opensearch_conn_id: opensearch connection to use
+ """
+
+ def __init__(
+ self,
+ *,
+ index_name: str,
+ index_body: dict[str, Any],
+ opensearch_conn_id: str = "opensearch_default",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.index_name = index_name
+ self.index_body = index_body
+ self.opensearch_conn_id = opensearch_conn_id
+
+ @cached_property
+ def hook(self) -> OpenSearchHook:
+ """Gets an instance of an OpenSearchHook."""
+ return OpenSearchHook(open_search_conn_id=self.opensearch_conn_id, log_query=False)
+
+ def execute(self, context: Context) -> Any:
+ """Creates an index on an Open Search cluster."""
+ try:
+ self.hook.get_client.indices.create(index=self.index_name, body=self.index_body)
+ except Exception as e:
+ raise AirflowException(e)
+
+
+class OpenSearchAddDocumentOperator(BaseOperator):
+ """
+ Adds a new document to a given Index. It will either add or overwrite an existing document.
Review Comment:
```suggestion
Add a new document to a given Index or overwrite an existing one.
```
##########
docs/apache-airflow-providers-opensearch/connections/opensearch.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+
+OpenSearch Connection
+=====================
+The Open Search connection provides credentials for an OpenSearch instance.
+
+Configuring the Connection
+--------------------------
+Host (required)
+ The host address of the Open Search instance.
+Login (required)
+ The login user.
+Password (required)
+ The password for the login user.
+Extra (optional)
+ Specifying the extra parameters as a (json dictionary) that can be used in the Open Search connection.
Review Comment:
Should this be OpenSearch? Or elsewhere you used `opensearch connection` so maybe that is the right use here? Not sure. Branding and naming issues are always tricky, I'm not sure how the OS project handles their name in use like this.
##########
docs/apache-airflow-providers-opensearch/operators/opensearch.rst:
##########
@@ -0,0 +1,72 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+=============
+OpenSearch
+=============
+
+`OpenSearch <https://opensearch.org/>`__
+
+
+Operators
+---------
+
+.. _howto/operator:OpenSearchCreateIndexOperator:
+
+Create an Index in Open Search
+==============================
+
+Use the :class:`~airflow.providers.opensearch.operators.opensearch.OpenSearchCreateIndexOperator`
+to create a new index in an Open Search domain.
+
+
+
+.. exampleinclude:: /../../tests/system/providers/opensearch/example_opensearch.py
+ :language: python
+ :start-after: [START howto_operator_opensearch_create_index]
+ :dedent: 4
+ :end-before: [END howto_operator_opensearch_create_index]
+
+
+.. _howto/operator:OpenSearchAddDocumentOperator:
+
+Add a Document to an Index on OpenSearch
+=========================================
+
+Use the :class:`~airflow.providers.opensearch.operators.opensearch.OpenSearchAddDocumentOperator`
+to add single documents to an Open Search Index
Review Comment:
Check the generated output but I think when this gets parsed into an actual doc file this is going to read awkwardly. In the AWS package we say "use :class:`~airflow.providers.opensearch.operators.opensearch.OpenSearchAddDocumentOperator` ..." because "use the..." sounded a little odd when the link got parsed. I'll leave the call up to you, just a suggestion.
--
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
Re: [PR] Add Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1762995522
If the checks fail after this merge from main do you want me to include any fixes here for the breeze mypy issue? I think it just needs to be changed from ignore[misc] to ignore[attr-defined]. Not sure about the unit tests that are failing though from core.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1349796246
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ Provide a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def conn(self):
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
Review Comment:
There was some miscommunication via the thread on the branches showing up from creating this branch not from mainline. So I tried to be proactive and drop those commits from this branch which ideally I don’t want to push that change unless it was absolutely necessary. I can clean this up if it is blocking, the suggestion you have can’t be applied given the references to this method in other places in the code.
##########
airflow/providers/opensearch/hooks/opensearch.py:
##########
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from functools import cached_property
+from typing import Any
+
+from opensearchpy import OpenSearch, RequestsHttpConnection
+
+from airflow.exceptions import AirflowException
+from airflow.hooks.base import BaseHook
+
+
+class OpenSearchHook(BaseHook):
+ """
+ Provide a thin wrapper around the OpenSearch client.
+
+ :param: open_search_conn_id: Connection to use with Open Search
+ :param: log_query: Whether to log the query used for Open Search
+ """
+
+ conn_name_attr = "opensearch_conn_id"
+ default_conn_name = "opensearch_default"
+ conn_type = "opensearch"
+ hook_name = "OpenSearch Hook"
+
+ def __init__(self, open_search_conn_id: str, log_query: bool, **kwargs: Any):
+ super().__init__(**kwargs)
+ self.conn_id = open_search_conn_id
+ self.log_query = log_query
+
+ self.use_ssl = self.conn.extra_dejson.get("use_ssl", False)
+ self.verify_certs = self.conn.extra_dejson.get("verify_certs", False)
+ self.__SERVICE = "es"
+
+ @cached_property
+ def conn(self):
+ return self.get_connection(self.conn_id)
+
+ @cached_property
+ def get_client(self) -> OpenSearch:
Review Comment:
There was some miscommunication via the thread on the branches showing up from creating this branch not from mainline. So I tried to be proactive and drop those commits from this branch which ideally I don’t want to push that change unless it was absolutely necessary. I can clean this up if it is blocking, the suggestion you have can’t be applied given the references to this method in other places in the code.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1346600005
##########
docs/apache-airflow-providers-opensearch/operators/opensearch.rst:
##########
@@ -0,0 +1,72 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+=============
+OpenSearch
+=============
+
+`OpenSearch <https://opensearch.org/>`__
+
+
+Operators
+---------
+
+.. _howto/operator:OpenSearchCreateIndexOperator:
+
+Create an Index in Open Search
+==============================
+
+Use the :class:`~airflow.providers.opensearch.operators.opensearch.OpenSearchCreateIndexOperator`
+to create a new index in an Open Search domain.
+
+
+
+.. exampleinclude:: /../../tests/system/providers/opensearch/example_opensearch.py
+ :language: python
+ :start-after: [START howto_operator_opensearch_create_index]
+ :dedent: 4
+ :end-before: [END howto_operator_opensearch_create_index]
+
+
+.. _howto/operator:OpenSearchAddDocumentOperator:
+
+Add a Document to an Index on OpenSearch
+=========================================
+
+Use the :class:`~airflow.providers.opensearch.operators.opensearch.OpenSearchAddDocumentOperator`
+to add single documents to an Open Search Index
Review Comment:
Fixed.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1746106253
For question number 2 on the PR when originally cut, the only path forward it seems with mypy was to use the Any type. Still open to other ideas if there is a way to be more specific with the type for the Operators that take high level objects for the high level client implementation of OpenSearch.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1343689432
##########
airflow/providers/opensearch/provider.yaml:
##########
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-opensearch
+name: Opensearch
+description: |
+ `Opensearch <https://opensearch.org/>`__
+
+suspended: false
+versions:
+ - 1.0.0
+
+dependencies:
+ - apache-airflow>=2.4.3
Review Comment:
lets set 2.5.0
all provider releases after 19th of September 2023 will have 2.5.0
https://github.com/apache/airflow/blob/main/PROVIDERS.rst#id5
##########
generated/provider_dependencies.json:
##########
@@ -28,6 +28,10 @@
"boto3>=1.28.0",
"botocore>=1.31.0",
"jsonpath_ng>=1.5.3",
+ "mypy-boto3-appflow>=1.24.0",
+ "mypy-boto3-rds>=1.24.0",
+ "mypy-boto3-redshift-data>=1.24.0",
+ "mypy-boto3-s3>=1.24.0",
Review Comment:
I wonder where did this come from?
this file is auto generated by pre-commits.
maybe you used another version of provider.yaml when you run the tests?
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347718163
##########
tests/providers/opensearch/operators/test_opensearch.py:
##########
@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from opensearchpy import Document, Keyword, Text
+
+from airflow.models import DAG
+from airflow.providers.opensearch.operators.opensearch import (
+ OpenSearchAddDocumentOperator,
+ OpenSearchCreateIndexOperator,
+ OpenSearchQueryOperator,
+)
+from airflow.utils.timezone import datetime
+
+TEST_DAG_ID = "unit_tests"
+DEFAULT_DATE = datetime(2018, 1, 1)
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class FakeDocument(Document):
+ title = Text(fields={"raw": Keyword()})
+ author = Text()
+ published = Text()
+
+ def save(self, **kwargs):
+ return super().save(**kwargs)
+
+
+class TestOpenSearchQueryOperator:
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
+
+ self.open_search = OpenSearchQueryOperator(
+ task_id="test_opensearch_query_operator",
+ index_name="test_index",
+ query={
+ "size": 5,
+ "query": {"multi_match": {"query": "test", "fields": ["test_title^2", "test_type"]}},
+ },
+ )
+
+ def test_init(self):
+ assert self.open_search.task_id == "test_opensearch_query_operator"
+ assert self.open_search.opensearch_conn_id == "opensearch_default"
+ assert self.open_search.query["size"] == 5
+
+ def test_search_query(self, mock_hook):
+ result = self.open_search.execute({})
+ assert result == MOCK_SEARCH_RETURN
+
+
+class TestOpenSearchCreateIndexOperator:
+ # This test does not test execute logic because there is only a redirect to the OpenSearch
+ # client.
+ def setup_method(self):
+ args = {
+ "owner": "airflow",
+ "start_date": DEFAULT_DATE,
+ }
+
+ self.dag = DAG(f"{TEST_DAG_ID}test_schedule_dag_once", default_args=args, schedule="@once")
Review Comment:
You might consider moving the dag creation into a fixture since all these tests are repeating that block. In addition, as a suggestion based on personal preference, I'd also inline the args. There doesn't appear to be any reason to have them stored as a variable 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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347706512
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
+ result = OpenSearchHook(open_search_conn_id="opensearch_default", log_query=True).search(
+ index_name="testIndex",
+ query={"size": 1, "query": {"multi_match": {"query": "test", "fields": ["testField"]}}},
+ )
Review Comment:
[Here and elsewhere] Functionally identical but perhaps a little easier to read if you split the hook out into its own assignment? Not sure, I'll leave it up to you
```suggestion
hook = OpenSearchHook(
open_search_conn_id="opensearch_default",
log_query=True,
)
result = hook.search(
index_name="testIndex",
query={"size": 1, "query": {"multi_match": {"query": "test", "fields": ["testField"]}}},
)
```
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347707573
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
Yes this works. See https://docs.pytest.org/en/6.2.x/fixture.html#scope-sharing-fixtures-across-classes-modules-packages-or-session
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347704163
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
and the fixture doesn't have autouse=True.... I wouldn't expect this to work as written
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on code in PR #34705:
URL: https://github.com/apache/airflow/pull/34705#discussion_r1347718577
##########
tests/providers/opensearch/hooks/test_opensearch.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.opensearch.hooks.opensearch import OpenSearchHook
+
+MOCK_SEARCH_RETURN = {"status": "test"}
+
+
+class TestOpenSearchHook:
+ def test_hook_search(self, mock_hook):
Review Comment:
autouse=True would mean that tests do not have to explicitly request the fixture by having it defined as a parameter for the test function. I am just being explicit by requesting the fixture but I could switch it to autouse.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "cjames23 (via GitHub)" <gi...@apache.org>.
cjames23 commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1749774067
@ferruzzi those should not have any impact. I think they showed up because I didn’t pay attention when creating this new branch as my local was still on the chime notifier branch. I can try to cherry pick them out if there is a blocking concern but all of that code was already merged.
--
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
Re: [PR] Open Search Provider [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #34705:
URL: https://github.com/apache/airflow/pull/34705#issuecomment-1751018425
> I think would be easier to just squash all commits
Yeah, given that there is no impact, I'll agree there.
--
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