You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mahammi (via GitHub)" <gi...@apache.org> on 2023/07/10 18:13:56 UTC
[GitHub] [airflow] mahammi opened a new pull request, #32498: Add Eventbridge PutEvents operator and hook
mahammi opened a new pull request, #32498:
URL: https://github.com/apache/airflow/pull/32498
Adding EventBridge hook and PutEvents operator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1634684246
I wrote a message about this last night, I wonder what happened to it..... strange.
The failing test has a suggestion to add the operator to a "no example available" list [here](https://github.com/apache/airflow/blob/main/tests/always/test_project_structure.py#L284), I think that makes more sense than a non-functional dag for 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
[GitHub] [airflow] mahammi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258707589
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
Review Comment:
Since PutEvents allows for multiple custom events to be sent, it is possible that some may have been sent to EventBridge successfully while others failed. Here I have decided to log the ones that have failed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] o-nikolas commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1634687942
> The failing test has a suggestion to add the operator to a "no example available" list [here](https://github.com/apache/airflow/blob/main/tests/always/test_project_structure.py#L284), I think that makes more sense than a non-functional dag for now.
I'd be worried about setting a precedent doing this. People will start ignoring them and forgetting about it.
What's the concern with writing a functioning system test? It's a pretty well dialed process nowadays.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Adaverse commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "Adaverse (via GitHub)" <gi...@apache.org>.
Adaverse commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258685917
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
+
+ raise AirflowException(
+ f"{response['FailedEntryCount']} entries in this request have failed to send."
Review Comment:
Do we need to log the error since we are throwing an exception with details about it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1633301026
@eladkal @hussein-awala @Adaverse
Hey all, I have made some edits according to your comments. Feel free to review when you have a moment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Adaverse commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "Adaverse (via GitHub)" <gi...@apache.org>.
Adaverse commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258709941
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
+
+ raise AirflowException(
+ f"{response['FailedEntryCount']} entries in this request have failed to send."
Review Comment:
Actually I was talking about this line,
```
self.log.error("Some events have failed to send.")
```
Note to self - point to the exact line in review. MB
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1264209408
##########
tests/system/providers/amazon/aws/example_eventbridge.py:
##########
@@ -0,0 +1,54 @@
+# 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 datetime import datetime
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.eventbridge import EventBridgePutEventsOperator
+
+DAG_ID = "example_eventbridge"
+ENTRIES = [
+ {
+ "Detail": '{"event-name": "custom-event"}',
+ "EventBusName": "custom-bus",
+ "Source": "example.myapp",
+ "DetailType": "Sample Custom Event",
+ }
+]
+
+with DAG(
+ dag_id=DAG_ID,
+ schedule="@once",
+ start_date=datetime(2021, 1, 1),
+ tags=["example"],
+ catchup=False,
+) as dag:
+
+ # [START howto_operator_eventbridge_put_events]
+
+ put_events = EventBridgePutEventsOperator(
+ task_id="put_events_task", entries=ENTRIES, region_name="us-east-1"
+ )
+
+ # [END howto_operator_eventbridge_put_events]
+
Review Comment:
When you add another task you'll also need to add the watcher (see other system tests for examples) but I think this should be fine for now.
##########
tests/system/providers/amazon/aws/example_eventbridge.py:
##########
@@ -0,0 +1,54 @@
+# 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 datetime import datetime
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.eventbridge import EventBridgePutEventsOperator
+
+DAG_ID = "example_eventbridge"
+ENTRIES = [
+ {
+ "Detail": '{"event-name": "custom-event"}',
+ "EventBusName": "custom-bus",
+ "Source": "example.myapp",
+ "DetailType": "Sample Custom Event",
+ }
+]
+
+with DAG(
+ dag_id=DAG_ID,
+ schedule="@once",
+ start_date=datetime(2021, 1, 1),
+ tags=["example"],
+ catchup=False,
+) as dag:
+
+ # [START howto_operator_eventbridge_put_events]
+
+ put_events = EventBridgePutEventsOperator(
+ task_id="put_events_task", entries=ENTRIES, region_name="us-east-1"
Review Comment:
I think we can drop the region name. We want to keep these as minimal as possible since they are getting embedded in docs and region_name gets a default value in the operator, so it's not required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Adaverse commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "Adaverse (via GitHub)" <gi...@apache.org>.
Adaverse commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258688520
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
Review Comment:
I think this comment may not be required since the code snippet is very clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] o-nikolas commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1262818421
##########
docs/apache-airflow-providers-amazon/operators/events.rst:
##########
@@ -0,0 +1,47 @@
+ .. 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.
+
+========================================
+Amazon EventBridge
+========================================
+
+`Amazon Eventbridge <https://docs.aws.amazon.com/eventbridge/>`__ is a serverless event bus service that makes it easy
+to connect your applications with data from a variety of sources. EventBridge delivers a stream of real-time data from
+your own applications, software-as-a-service (SaaS) applications, and AWS services and routes that data to targets such
+as AWS Lambda. You can set up routing rules to determine where to send your data to build application architectures
+that react in real time to all of your data sources. EventBridge enables you to build event-driven architectures
+that are loosely coupled and distributed.
+
+Prerequisite Tasks
+------------------
+
+.. include:: ../_partials/prerequisite_tasks.rst
+
+Operators
+-------
Review Comment:
There are some doc builds failures, including this line not matching the length of the word above it. You can run the docs build on your laptop for just the amazon provider like this: `breeze build-docs --package-filter apache-airflow-providers-amazon`.
Ensure that passes green :green_circle:
```suggestion
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
[GitHub] [airflow] o-nikolas commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1636099840
> I can make an example DAG, but I don't know how useful that would be at the moment. Once more operators are made, we can always edit the example DAG, though.
@mahammi yeah I think this is the best approach. Just iteratively build the test as we go. You don't have to write the CDK portion just yet, and the test wont run on our system test infra until you do. So that should all be fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1259055994
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
+
+ raise AirflowException(
+ f"{response['FailedEntryCount']} entries in this request have failed to send."
+ )
+
+ return [e["EventId"] for e in response["Entries"]]
Review Comment:
Interesting, I wasn't aware of that either. Nice catch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1634675854
> You're missing an example dag/system test for this new operator
Should we create a "broken dag" that just shows how to use the operator for now, and focus and making it a functioning system test later on ? Would that be a good course of action ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258707589
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
Review Comment:
Since PutEvents allows for multiple custom events to be sent, it is possible that some may have been sent to EventBridge successfully while others failed. Here I have decided to log the ones that have failed. With that being said - do you think it's fair to parse the response?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] hussein-awala commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258922424
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
Review Comment:
```suggestion
from typing import TYPE_CHECKING, Sequence
```
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
Review Comment:
```suggestion
from functools import cached_property
```
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
Review Comment:
```suggestion
if TYPE_CHECKING:
from airflow.utils.context import Context
```
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
Review Comment:
I agree, we should only log events with `ErrorCode`
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
Review Comment:
From AWS API doc, we could have this repsonse:
```json
{
"FailedEntryCount": 0,
"Entries": [
{
"EventId": "11710aed-b79e-4468-a20b-bb3c0c3b4860"
},
{
"EventId": "d804d26a-88db-4b66-9eaf-9a11c708ae82"
}
]
}
```
so you check is not valid, instead you can use:
```suggestion
if response.get("FailedEntryCount"):
```
##########
tests/providers/amazon/aws/operators/test_eventbridge.py:
##########
@@ -0,0 +1,73 @@
+# 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 unittest import mock
+from unittest.mock import MagicMock
+
+import pytest
+
+from airflow import AirflowException
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.operators.eventbridge import EventBridgePutEventsOperator
+
+TASK_ID = "events_putevents_job"
+ENTRIES = [{"Detail": "test-detail", "Source": "test-source", "DetailType": "test-detail-type"}]
+FAILED_ENTRIES_RESPONSE = [{"ErrorCode": "test_code"}, {"ErrorCode": "test_code"}]
+
+
+class TestEventBridgePutEventsOperator:
+ def test_init(self):
+ operator = EventBridgePutEventsOperator(
+ task_id=TASK_ID,
+ entries=ENTRIES,
+ )
+
+ assert operator.entries == ENTRIES
+
+ @mock.patch.object(EventBridgeHook, "conn")
+ def test_execute(self, mock_conn: MagicMock):
+ hook_response = {"Entries": [{"EventId": "foobar"}]}
Review Comment:
not a valid response:
```suggestion
hook_response = {"FailedEntryCount": 0, "Entries": [{"EventId": "foobar"}]}
```
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
+
+ raise AirflowException(
+ f"{response['FailedEntryCount']} entries in this request have failed to send."
+ )
+
+ return [e["EventId"] for e in response["Entries"]]
Review Comment:
The task will fail when the number of entries exceeds `AIRFLOW__CORE__MAX_MAP_LENGTH` ([doc](https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#max-map-length)).
I suggest returning them when `do_xcom_push` is True (it's True by default, but at least the user will have an option to stop pushing xcom)
```suggestion
if self.do_xcom_push:
return [e["EventId"] for e in response["Entries"]]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal merged pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal merged PR #32498:
URL: https://github.com/apache/airflow/pull/32498
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258692161
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
Review Comment:
I think this should be
`from functools import cached_property`
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
Review Comment:
Shouldn't this be under `TYPE_CHECKING` ?
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
Review Comment:
Shouldn't we just dump the response to the log?
Why do we need to parse it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1636572812
@Adaverse - Anything else to add?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1266273467
##########
docs/apache-airflow-providers-amazon/operators/events.rst:
##########
Review Comment:
Lets keep naming convention.
It should be `eventbridge.rst`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1259031135
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
Review Comment:
Thanks for catching this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258707062
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
+
+ raise AirflowException(
+ f"{response['FailedEntryCount']} entries in this request have failed to send."
Review Comment:
The reason for logging the errors from the response is because the API responds with an error code and error message. This would give the user more context as to why the failure(s) has occurred.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on a diff in pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on code in PR #32498:
URL: https://github.com/apache/airflow/pull/32498#discussion_r1258707589
##########
airflow/providers/amazon/aws/operators/eventbridge.py:
##########
@@ -0,0 +1,87 @@
+# 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 Sequence
+
+from airflow import AirflowException
+from airflow.decorators.base import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.eventbridge import EventBridgeHook
+from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.utils.context import Context
+
+
+class EventBridgePutEventsOperator(BaseOperator):
+ """
+ Put Events onto Amazon EventBridge.
+
+ :param entries: the list of events to be put onto EventBridge, each event is a dict (required)
+ :param endpoint_id: the URL subdomain of the endpoint
+ :param aws_conn_id: the AWS connection to use
+ :param region_name: the region where events are to be sent
+
+ """
+
+ template_fields: Sequence[str] = ("entries", "endpoint_id", "aws_conn_id", "region_name")
+
+ def __init__(
+ self,
+ *,
+ entries: list[dict],
+ endpoint_id: str | None = None,
+ aws_conn_id: str = "aws_default",
+ region_name: str | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.entries = entries
+ self.endpoint_id = endpoint_id
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+
+ @cached_property
+ def hook(self) -> EventBridgeHook:
+ """Create and return an EventBridgeHook."""
+ return EventBridgeHook(aws_conn_id=self.aws_conn_id, region_name=self.region_name)
+
+ def execute(self, context: Context):
+
+ response = self.hook.conn.put_events(
+ **trim_none_values(
+ {
+ "Entries": self.entries,
+ "EndpointId": self.endpoint_id,
+ }
+ )
+ )
+
+ self.log.info("Sent %d events to EventBridge.", len(self.entries))
+
+ # If events have failed, log those error codes and messages to console, and raise an exception.
+
+ if "FailedEntryCount" in response:
+ self.log.error("Some events have failed to send.")
+ for event in response["Entries"]:
+ if "ErrorCode" in event:
+ self.log.error(event)
Review Comment:
Since PutEvents allows for multiple custom events to be sent, it is possible that some may have been sent to EventBridge successfully while other failed. Here I have decided to log the ones that have failed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1643406455
Thank you @mahammi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mahammi commented on pull request #32498: Add Eventbridge PutEvents operator and hook
Posted by "mahammi (via GitHub)" <gi...@apache.org>.
mahammi commented on PR #32498:
URL: https://github.com/apache/airflow/pull/32498#issuecomment-1634716859
> > The failing test has a suggestion to add the operator to a "no example available" list [here](https://github.com/apache/airflow/blob/main/tests/always/test_project_structure.py#L284), I think that makes more sense than a non-functional dag for now.
>
> I'd be worried about setting a precedent doing this. People will start ignoring them and forgetting about it.
>
> What's the concern with writing a functioning system test? It's a pretty well dialed process nowadays.
> > The failing test has a suggestion to add the operator to a "no example available" list [here](https://github.com/apache/airflow/blob/main/tests/always/test_project_structure.py#L284), I think that makes more sense than a non-functional dag for now.
>
> I'd be worried about setting a precedent doing this. People will start ignoring them and forgetting about it.
>
> What's the concern with writing a functioning system test? It's a pretty well dialed process nowadays.
The main concern that I discussed with Dennis is just the fact that there isn't a way to see whether events have been sent to EventBridge. All events are either sent to the targets via rules, or dropped if there are no targets immediately after receiving the events, meaning we would need the PutRule operator before we can do much with this one (in a DAG at least). I can make an example DAG, but I don't know how useful that would be at the moment. Once more operators are made, we can always edit the example DAG, though.
--
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