You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Taragolis (via GitHub)" <gi...@apache.org> on 2023/10/05 19:16:33 UTC
[PR] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Taragolis opened a new pull request, #34784:
URL: https://github.com/apache/airflow/pull/34784
<!--
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/
-->
Continuation #24246
This base operators and sensors might help to resolve inconsistency in multiple operators and sensors:
1. Some of current sensors/operators accept only `aws_conn_id`
2. Some of current sensors/operators accept `region` instead of `region_name`
In additional this implementation propagate additional parameters into the hook:
1. Operator/Sensor `botocore_config` -> Hook `config`
2. Operator/Sensor `verify` -> Hook `verify`
> [!NOTE]
> `AwsBaseSensor` created by "old methodic"
> 1. Copy code from `AwsBaseOperator`
> 2. Paste in new module
>
> I've tried to create Mixin class however it not works well breaks something in BaseOperator
> If someone know how to implements it by DRY, without use additional Metaclasses and/or class decorators feel free to suggest.
<!-- 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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348077800
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Yeah, I think you are right. Leave it as you have it, I guess. Thanks for considering it 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
Re: [PR] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348066197
##########
airflow/providers/amazon/aws/hooks/base_aws.py:
##########
@@ -463,15 +463,17 @@ def __init__(
region_name: str | None = None,
client_type: str | None = None,
resource_type: str | None = None,
- config: Config | None = None,
+ config: Config | dict[str, Any] | None = None,
) -> None:
super().__init__()
self.aws_conn_id = aws_conn_id
self.client_type = client_type
self.resource_type = resource_type
self._region_name = region_name
- self._config = config
+ if config and isinstance(config, dict):
+ config = Config(**config)
+ self._config = config or None
Review Comment:
A bit dumb check, however it required to pass tests, some of AWS test patch `isinstance` as result `isinstance(None, dict)` evaluate as True
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347898473
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Could we make BaseSensor inherit BaseOperator and save a lot of duplicate code? A Sensor is just a specialized Operator, really.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348316001
##########
tests/always/test_project_structure.py:
##########
@@ -403,6 +403,9 @@ class TestAmazonProviderProjectStructure(ExampleCoverageTest):
"airflow.providers.amazon.aws.transfers.exasol_to_s3.ExasolToS3Operator",
# Glue Catalog sensor difficult to test
"airflow.providers.amazon.aws.sensors.glue_catalog_partition.GlueCatalogPartitionSensor",
+ # Base Classes
+ "airflow.providers.amazon.aws.operators.base_aws.AwsBaseOperator",
+ "airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor",
Review Comment:
You need to move these to the `BASE_CLASSES`
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348746185
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
Review Comment:
Yep, both are good to me :+1:
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1353229032
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
Review Comment:
> It is also might be a good idea to make this classes internal only.
+1
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348927935
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
Review Comment:
I had to research that one too, I have never seen that in use. It's kinda handy.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347991914
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
Review Comment:
And wondering about module naming because I a bit confused:
Suggestion in old PR use **base_aws**: https://github.com/apache/airflow/pull/24246#discussion_r892775749
Naming for Base Trigger: airflow.providers.amazon.aws.triggers.**base**.AwsBaseWaiterTrigger
cc @ferruzzi @vincbeck @o-nikolas
> There are only two hard things in Computer Science: cache invalidation and naming things
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1353234785
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
Review Comment:
Using the default AWS Connection id `aws_default` (as long as it is created and remains untouched by the user) also uses the default boto3 behaviour. Might be worth mentioning.
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,97 @@
+# 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.models import BaseOperator
+from airflow.providers.amazon.aws.utils.mixins import AwsBaseHookMixin, AwsHookParams, AwsHookType
+
+
+class AwsBaseOperator(BaseOperator, AwsBaseHookMixin[AwsHookType]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ .. warning::
+ Only for internal usage, this class might be changed, renamed or removed in the future
+ without any further notice.
Review Comment:
Regarding this, should we change the class name to `_AwsBaseOperator`?
Same for the Sensor and mixin classes
##########
airflow/providers/amazon/aws/utils/mixins.py:
##########
@@ -0,0 +1,165 @@
+# 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.
+
+"""
+This module contains different internal mixin classes for internal of Amazon Provider.
Review Comment:
```suggestion
This module contains different mixin classes for internal use within the Amazon provider.
```
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347884136
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
Review Comment:
I think this is repeating itself, but maybe I'm missing something.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347948429
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
This might works
```python
class AwsBaseSensor(AwsBaseOperator, BaseSensorOperator)
...
```
However we would step into [MRO](https://www.python.org/download/releases/2.3/mro/) field, if we do something wrong and forget about some important thing, it might easily cause an issue in multiple sensor eventually.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347986606
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Just for comparison, seems like it should work `BaseSensorOperator` use before `BaseOperator` however if it not strict concern to DRY and use this kind of multi inheritance I would rather keep it with bunch of repeatable code. At least until we do not move existed AWS Sensors to AwsBaseSensor
MRO: class AwsBaseSensor(AwsBaseOperator, BaseSensorOperator)
```
1: <class 'airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor'>
2: <class 'airflow.providers.amazon.aws.operators.base_aws.AwsBaseOperator'>
3: <class 'airflow.sensors.base.BaseSensorOperator'>
4: <class 'airflow.models.baseoperator.BaseOperator'>
5: <class 'airflow.models.abstractoperator.AbstractOperator'>
6: <class 'airflow.template.templater.Templater'>
7: <class 'airflow.models.skipmixin.SkipMixin'>
8: <class 'airflow.utils.log.logging_mixin.LoggingMixin'>
9: <class 'airflow.models.taskmixin.DAGNode'>
10: <class 'airflow.models.taskmixin.DependencyMixin'>
11: <class 'typing.Generic'>
12: <class 'object'>
```
MRO: class AwsBaseSensor(BaseSensorOperator, Generic[_AwsHook]):
```
1: <class 'airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor'>
2: <class 'airflow.sensors.base.BaseSensorOperator'>
3: <class 'airflow.models.baseoperator.BaseOperator'>
4: <class 'airflow.models.abstractoperator.AbstractOperator'>
5: <class 'airflow.template.templater.Templater'>
6: <class 'airflow.models.skipmixin.SkipMixin'>
7: <class 'airflow.utils.log.logging_mixin.LoggingMixin'>
8: <class 'airflow.models.taskmixin.DAGNode'>
9: <class 'airflow.models.taskmixin.DependencyMixin'>
10: <class 'typing.Generic'>
11: <class 'object'>
```
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348746923
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
Review Comment:
I see, thanks for the explanation
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347878899
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
Review Comment:
This changes also required to make changes in `airflow.providers.amazon.aws.triggers.base.AwsBaseWaiterTrigger` as follow-up.
That is also why I make accept `botocore_config` as dictionary and not `botocore.config.Config` object
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348730693
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
Review Comment:
Just for skip previous traceback, e.g. if developer set `aws_hook_class` then it failed with TypeError on `issubclass` and traceback will contain information about error which raised from `issubclass`
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347889605
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
All of the above suggestions apply down here as well.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347868314
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
Review Comment:
It is also might be a good idea to make this classes internal only.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1353473369
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,97 @@
+# 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.models import BaseOperator
+from airflow.providers.amazon.aws.utils.mixins import AwsBaseHookMixin, AwsHookParams, AwsHookType
+
+
+class AwsBaseOperator(BaseOperator, AwsBaseHookMixin[AwsHookType]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ .. warning::
+ Only for internal usage, this class might be changed, renamed or removed in the future
+ without any further notice.
Review Comment:
There is kind of tradeoff between common Python's agreement about internal variables/classes/methods and airflow public models. `_` shows that is internal stuff however IDE doesn't like this imports even with the same package (I guess static checks is fine) in the other hand we have some methods and class in core which do no prefixed by `_` but have a description and `:meta private:`.
There is also maybe a good idea to make this stuff public after a while for create custom operators, but for now I would rather keep it internal only
In the end no one could stop end users to use this stuff directly
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1353460934
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
Review Comment:
Maybe we could try to rewrite it to something shorter with reference to AWS Connection, like we have in other providers, e.g. Slack
https://github.com/apache/airflow/blob/ed6a4fd1162407a2007d949765eab5749af1c3ac/airflow/providers/slack/operators/slack_webhook.py#L42-L43
Example in Auto API:
https://airflow.apache.org/docs/apache-airflow-providers-slack/8.1.0/_api/airflow/providers/slack/operators/slack_webhook/index.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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1347884136
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
Review Comment:
This repeating itself, but maybe I'm missing something.
##########
tests/providers/amazon/aws/operators/test_base_aws.py:
##########
@@ -0,0 +1,132 @@
+# 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 pytest
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.hooks.base import BaseHook
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
+
+TEST_CONN = "aws_test_conn"
+
+
+class FakeS3Hook(AwsBaseHook):
+ """Hook for tests, implements thin-wrapper around s3 client."""
+
+ def __init__(self, **kwargs):
+ kwargs.update({"client_type": "s3", "resource_type": None})
+ super().__init__(**kwargs)
+
+
+class FakeS3Operator(AwsBaseOperator):
+ aws_hook_class = FakeS3Hook
+
+
+@pytest.fixture(autouse=True)
+def fake_conn(monkeypatch):
+ monkeypatch.setenv(f"AWS_CONN_{TEST_CONN.upper()}", '{"conn_type": "aws"}')
+
+
+class TestAwsBaseOperator:
+ def test_default_parameters(self):
+ op = FakeS3Operator(task_id="fake_task_id")
+ msg = "Attention! Changes in default parameters might produce breaking changes in multiple operators"
Review Comment:
I like this one! Nice addition.
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` properly set with
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
+
+ if region := kwargs.pop("region", None):
+ warnings.warn(_REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
+ if region_name and region_name != region:
+ raise AirflowException(
+ f"Ambiguous `region_name` provided, region_name={region_name!r}, region={region!r}."
+ )
+ region_name = region
+
+ super().__init__(**kwargs)
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+ self.verify = verify
+ self.botocore_config = botocore_config
+
+ @property
+ def _hook_parameters(self) -> dict[str, Any]:
+ """Mapping parameters for build boto3-related hook.
+
+ Only required to be overwritten for thick-wrapped Hooks.
+ """
+ return {
+ "aws_conn_id": self.aws_conn_id,
+ "region_name": self.region_name,
+ "verify": self.verify,
+ "config": self.botocore_config,
+ }
+
+ @cached_property
+ @final
+ def hook(self) -> _AwsHook:
+ """
+ Return AWS Provider's hook based on ``aws_hook_class``.
+
+ This method implementation should be taken as a final, which a good for
+ thin-wrapped Hooks around boto3, for thick-wrapped Hooks developer
+ should consider to overwrite ``_hook_parameters`` method instead.
+ """
+ return self.aws_hook_class(**self._hook_parameters)
+
+ @property
+ @final
+ def region(self) -> str | None:
+ """Alias for ``region_name``, uses for compatibility (deprecated)."""
Review Comment:
```suggestion
"""Alias for ``region_name``, used for compatibility (deprecated)."""
```
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
Review Comment:
```suggestion
"""
Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
```
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` properly set with
Review Comment:
```suggestion
# Validate if ``aws_hook_class`` is properly set.
```
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` properly set with
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
+
+ if region := kwargs.pop("region", None):
+ warnings.warn(_REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
+ if region_name and region_name != region:
+ raise AirflowException(
+ f"Ambiguous `region_name` provided, region_name={region_name!r}, region={region!r}."
+ )
+ region_name = region
+
+ super().__init__(**kwargs)
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+ self.verify = verify
+ self.botocore_config = botocore_config
+
+ @property
+ def _hook_parameters(self) -> dict[str, Any]:
+ """Mapping parameters for build boto3-related hook.
Review Comment:
```suggestion
"""
Mapping parameters to build boto3-related hooks.
```
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` properly set with
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
+
+ if region := kwargs.pop("region", None):
+ warnings.warn(_REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
+ if region_name and region_name != region:
+ raise AirflowException(
+ f"Ambiguous `region_name` provided, region_name={region_name!r}, region={region!r}."
Review Comment:
```suggestion
f"Conflicting `region_name` provided, region_name={region_name!r}, region={region!r}."
```
##########
tests/providers/amazon/aws/operators/test_base_aws.py:
##########
@@ -0,0 +1,132 @@
+# 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 pytest
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.hooks.base import BaseHook
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
+
+TEST_CONN = "aws_test_conn"
+
+
+class FakeS3Hook(AwsBaseHook):
+ """Hook for tests, implements thin-wrapper around s3 client."""
+
+ def __init__(self, **kwargs):
+ kwargs.update({"client_type": "s3", "resource_type": None})
+ super().__init__(**kwargs)
+
+
+class FakeS3Operator(AwsBaseOperator):
+ aws_hook_class = FakeS3Hook
+
+
+@pytest.fixture(autouse=True)
+def fake_conn(monkeypatch):
+ monkeypatch.setenv(f"AWS_CONN_{TEST_CONN.upper()}", '{"conn_type": "aws"}')
+
+
+class TestAwsBaseOperator:
+ def test_default_parameters(self):
+ op = FakeS3Operator(task_id="fake_task_id")
+ msg = "Attention! Changes in default parameters might produce breaking changes in multiple operators"
+ assert op.aws_conn_id == "aws_default", msg
+ assert op.region_name is None, msg
+ assert op.verify is None, msg
+ assert op.botocore_config is None, msg
+
+ def test_parameters(self):
+ op = FakeS3Operator(
+ task_id="fake-task-id",
+ aws_conn_id=TEST_CONN,
+ region_name="eu-central-1",
+ verify=False,
+ botocore_config={"read_timeout": 777, "connect_timeout": 42},
+ )
+
+ assert op.aws_conn_id == TEST_CONN
+ assert op.region_name == "eu-central-1"
+ assert op.verify is False
+ assert op.botocore_config == {"read_timeout": 777, "connect_timeout": 42}
+
+ hook = op.hook
+ assert isinstance(hook, FakeS3Hook)
+ assert hook.aws_conn_id == op.aws_conn_id
+ assert hook._region_name == op.region_name
+ assert hook._verify == op.verify
+ assert hook._config.read_timeout == 777
+ assert hook._config.connect_timeout == 42
+
+ @pytest.mark.parametrize(
+ "region, region_name",
+ [
+ pytest.param("eu-west-1", None, id="region-only"),
+ pytest.param("us-east-1", "us-east-1", id="non-ambiguous-params"),
+ ],
+ )
+ def test_deprecated_region_name(self, region, region_name):
+ warning_match = r"`region` is deprecated and will be removed"
+ with pytest.warns(AirflowProviderDeprecationWarning, match=warning_match):
+ op = FakeS3Operator(
+ task_id="fake-task-id",
+ aws_conn_id=TEST_CONN,
+ region=region,
+ region_name=region_name,
+ )
+ assert op.region_name == region
+
+ with pytest.warns(AirflowProviderDeprecationWarning, match=warning_match):
+ assert op.region == region
+
+ def test_ambiguous_region_name(self):
Review Comment:
If you accept the suggestion to reword the warning, you'll need to change this as well.
##########
airflow/providers/amazon/aws/sensors/base_aws.py:
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
All of he above suggestions apply down here as well.
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,150 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """Base AWS (Amazon) Operator Class for build operators in top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` properly set with
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
+
+ if region := kwargs.pop("region", None):
+ warnings.warn(_REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
+ if region_name and region_name != region:
+ raise AirflowException(
+ f"Ambiguous `region_name` provided, region_name={region_name!r}, region={region!r}."
+ )
+ region_name = region
+
+ super().__init__(**kwargs)
+ self.aws_conn_id = aws_conn_id
+ self.region_name = region_name
+ self.verify = verify
+ self.botocore_config = botocore_config
+
+ @property
+ def _hook_parameters(self) -> dict[str, Any]:
+ """Mapping parameters for build boto3-related hook.
+
+ Only required to be overwritten for thick-wrapped Hooks.
+ """
+ return {
+ "aws_conn_id": self.aws_conn_id,
+ "region_name": self.region_name,
+ "verify": self.verify,
+ "config": self.botocore_config,
+ }
+
+ @cached_property
+ @final
+ def hook(self) -> _AwsHook:
+ """
+ Return AWS Provider's hook based on ``aws_hook_class``.
+
+ This method implementation should be taken as a final, which a good for
+ thin-wrapped Hooks around boto3, for thick-wrapped Hooks developer
+ should consider to overwrite ``_hook_parameters`` method instead.
+ """
Review Comment:
```suggestion
This method implementation should be taken as a final for
thin-wrapped Hooks around boto3. For thick-wrapped Hooks developer
should consider to overwrite ``_hook_parameters`` method instead.
"""
```
May also need to adjust line breaks after this change.
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348641033
##########
tests/always/test_project_structure.py:
##########
@@ -403,6 +403,9 @@ class TestAmazonProviderProjectStructure(ExampleCoverageTest):
"airflow.providers.amazon.aws.transfers.exasol_to_s3.ExasolToS3Operator",
# Glue Catalog sensor difficult to test
"airflow.providers.amazon.aws.sensors.glue_catalog_partition.GlueCatalogPartitionSensor",
+ # Base Classes
+ "airflow.providers.amazon.aws.operators.base_aws.AwsBaseOperator",
+ "airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor",
Review Comment:
I was a bit blind 🤣
##########
tests/always/test_project_structure.py:
##########
@@ -403,6 +403,9 @@ class TestAmazonProviderProjectStructure(ExampleCoverageTest):
"airflow.providers.amazon.aws.transfers.exasol_to_s3.ExasolToS3Operator",
# Glue Catalog sensor difficult to test
"airflow.providers.amazon.aws.sensors.glue_catalog_partition.GlueCatalogPartitionSensor",
+ # Base Classes
+ "airflow.providers.amazon.aws.operators.base_aws.AwsBaseOperator",
+ "airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor",
Review Comment:
I was a bit blind 🤣
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348720686
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
Review Comment:
Why usually raise `ValueError` when input is incorrect. We can assume this is part of the input? Not strongly opinionated tho
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
Review Comment:
Out of curiosity. Why `from None`?
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
Review Comment:
Agree, naming is hard xD. I am fine with `base_aws.py`. It could be also:
- base_operator.py
- base.py
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
+
+ if region := kwargs.pop("region", None):
+ warnings.warn(_REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
+ if region_name and region_name != region:
+ raise AirflowException(
Review Comment:
Same here, `ValueError`?
--
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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #34784:
URL: https://github.com/apache/airflow/pull/34784#discussion_r1348727907
##########
airflow/providers/amazon/aws/operators/base_aws.py:
##########
@@ -0,0 +1,152 @@
+# 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 warnings
+from functools import cached_property
+from typing import Any, Generic, Sequence, TypeVar
+
+from typing_extensions import final
+
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
+_AwsHook = TypeVar("_AwsHook", bound=AwsGenericHook)
+_REGION_MSG = "`region` is deprecated and will be removed in the future. Please use `region_name` instead."
+
+
+class AwsBaseOperator(BaseOperator, Generic[_AwsHook]):
+ """
+ Base AWS (Amazon) Operator Class to build operators on top of AWS Hooks.
+
+ Examples:
+ .. code-block:: python
+
+ from airflow.providers.amazon.aws.hooks.foo_bar import FooBarThinHook, FooBarThickHook
+
+
+ class AwsFooBarOperator(AwsBaseOperator[FooBarThinHook]):
+ aws_hook_class = FooBarThinHook
+
+ def execute(self, context):
+ pass
+
+
+ class AwsFooBarOperator2(AwsBaseOperator[FooBarThickHook]):
+ aws_hook_class = FooBarThickHook
+
+ def __init__(self, *, spam: str, **kwargs):
+ super().__init__(**kwargs)
+ self.spam = spam
+
+ @property
+ def _hook_parameters(self):
+ return {**super()._hook_parameters, "spam": self.spam}
+
+ def execute(self, context):
+ pass
+
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is None or empty then the default boto3 behaviour is used. If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+ https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ aws_hook_class: type[_AwsHook]
+ template_fields: Sequence[str] = (
+ "aws_conn_id",
+ "region_name",
+ "botocore_config",
+ )
+
+ def __init__(
+ self,
+ *,
+ aws_conn_id: str | None = "aws_default",
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ):
+ # Validate if ``aws_hook_class`` is properly set.
+ if hasattr(self, "aws_hook_class"):
+ try:
+ if not issubclass(self.aws_hook_class, AwsGenericHook):
+ raise TypeError
+ except TypeError:
+ # Raise if ``aws_hook_class`` is not a class or not a subclass of Generic/Base AWS Hook
+ raise AirflowException(
+ f"Class attribute '{type(self).__name__}.aws_hook_class' "
+ f"is not a subclass of AwsGenericHook."
+ ) from None
+ else:
+ raise AirflowException(f"Class attribute '{type(self).__name__}.aws_hook_class' should be set.")
Review Comment:
Yeah, why not. But we could use `AttributeError` 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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on PR #34784:
URL: https://github.com/apache/airflow/pull/34784#issuecomment-1750665063
>I assume followup PR would be to change AWS operators/hooks to inherit from the base classes?
Yep
> We need also doc update to explain how to set the botocore parameters
I think it would be a good idea to add generic information here: https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/operators/index.html as soon as we change existed Operators, so them will have same sets of generic parameters:
- `aws_conn_id`: Operators which intends to use AwsBaseHook already have
- `region_name`: Some of operators use this parameter
- `verify`: Only one use right now
- `botocore_config`: None of of the operators use it right 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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on PR #34784:
URL: https://github.com/apache/airflow/pull/34784#issuecomment-1755464358
In [126a1f5](https://github.com/apache/airflow/pull/34784/commits/126a1f50178f1f9d2c603a08a483952bf126520e) split logic into two classes, for reduce repeatable code in `AwsBaseOperator` and `AwsBaseSensor` and in additional this might help to use in other places, e.g. Notifiers. Potentially it also could be use in `AwsBaseWaiterTrigger` however it use hook as a method, rather than 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] Implements `AwsBaseOperator` and `AwsBaseSensor` [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi merged PR #34784:
URL: https://github.com/apache/airflow/pull/34784
--
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