You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/02/19 17:23:31 UTC
[GitHub] [airflow] schirag1993 opened a new pull request #21686: add invoke lambda function operator
schirag1993 opened a new pull request #21686:
URL: https://github.com/apache/airflow/pull/21686
<!--
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 existing issue, reference it using one of the following:
closes: #ISSUE
related: [Related: https://github.com/apache/airflow/issues/21009](https://github.com/apache/airflow/issues/21009)
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ 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 change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+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 [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1054971901
@malthe, @edwardwang888, @ferruzzi, @mik-laj, would y'all also be able to provide further feedback/approval on the PR please? I hope the changes have addressed all existing comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815147817
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,99 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
Review comment:
```suggestion
aws_conn_id: str = 'aws_default',
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810687095
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
```suggestion
def invoke_lambda(
self,
*,
function_name: str,
invocation_type: Optional[str] = None,
log_type: Optional[str] = None,
client_context: Optional[str] = None,
payload: Optional[str] = None,
qualifier: Optional[str] = None,
):
```
How about make these all keyword-only arguments? This is probably how a user is supposed to use them anyway (the invocation line would not be readable otherwise), so let’s enforce them.
Same for other functions on the hook.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r811614220
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
+ for key in self.__dict__
+ if (
+ key
+ in [
+ "Payload",
+ "LogType",
+ "Qualifier",
+ "InvocationType",
+ "ClientContext",
+ ]
+ and self.__dict__[key] is not None
+ )
+ },
+ )
+ self.log.info("Lambda response metadata: %s", json.dumps(response.get("ResponseMetadata")))
Review comment:
@schirag1993 I think @malthe may have been referring to the same thing as https://github.com/apache/airflow/pull/21686#discussion_r810686375, which is already resolved?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r811614860
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
+ for key in self.__dict__
+ if (
+ key
+ in [
+ "Payload",
+ "LogType",
+ "Qualifier",
+ "InvocationType",
+ "ClientContext",
+ ]
+ and self.__dict__[key] is not None
+ )
+ },
+ )
+ self.log.info("Lambda response metadata: %s", json.dumps(response.get("ResponseMetadata")))
Review comment:
Yes, looks like it @edwardwang888. That should mark all comments addressed then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810756077
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
I agree @edwardwang888. Will make 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
[GitHub] [airflow] schirag1993 commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1051521174
> Sorry, I submitted those as single comments. Wish I could retroactively roll them into a review to reduce clutter.
It's no trouble at all, thank you for the review @ferruzzi. Allow me to work on these changes. It appears some tests are failing as well. Will work on the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810755989
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
Review comment:
+1, I'll make the required change @edwardwang888
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810755842
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
Hello @uranusjr, these were originally kwargs. However, based on @malthe's feedback (https://github.com/apache/airflow/pull/21686#discussion_r810589155) from an earlier discussion (which I agree with), it may be difficult for first timers to understand what is required and the typing info helps as well.
As we are building this for Airflow, it makes sense to spell out these parameters for users for better reference.
What are your thoughts? If the above does not make sense, I'll go ahead and make the required changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r811198322
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
Done @uranusjr! Thank you for the feedback. Today I learnt something new
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1053221198
Apologies for the trouble @potiuk. Working on the fixes today. Let me convert this into a draft.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815144081
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,99 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.function_name = function_name
+ self.payload = payload
+ self.log_type = log_type
+ self.qualifier = qualifier
+ self.invocation_type = invocation_type
+ self.client_context = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.function_name, self.payload)
+ response = hook.invoke_lambda(
+ function_name=self.function_name,
+ invocation_type=self.invocation_type,
+ log_type=self.log_type,
+ client_context=self.client_context,
+ payload=self.payload,
+ qualifier=self.qualifier,
+ )
+ self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
+ if response.get("StatusCode") not in success_status_codes:
+ raise ValueError('Lambda function did not execute', json.dumps(response.get("ResponseMetadata")))
+ if "FunctionError" in response:
+ error_payload_stream = response.get("Payload")
+ error_payload = error_payload_stream.read().decode()
+ raise ValueError(
+ 'Lambda function execution resulted in error',
+ {"ResponseMetadata": response.get("ResponseMetadata"), "Payload": error_payload},
+ )
+ self.log.info('Lambda function invocation succeeded: %r', response.get("ResponseMetadata"))
+ payload_stream = response.get("Payload")
+ payload = payload_stream.read().decode()
+ return payload
Review comment:
Seems like Lines 97/98 can be moved before the if "FunctionError" block and reused inside the block as well, no? You are getting and decoding the stream whether there is an error or not.
```suggestion
payload_stream = response.get("Payload")
payload = payload_stream.read().decode()
if "FunctionError" in response:
raise ValueError(
'Lambda function execution resulted in error',
{"ResponseMetadata": response.get("ResponseMetadata"), "Payload": payload},
)
self.log.info('Lambda function invocation succeeded: %r', response.get("ResponseMetadata"))
return payload
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815144081
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,99 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.function_name = function_name
+ self.payload = payload
+ self.log_type = log_type
+ self.qualifier = qualifier
+ self.invocation_type = invocation_type
+ self.client_context = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.function_name, self.payload)
+ response = hook.invoke_lambda(
+ function_name=self.function_name,
+ invocation_type=self.invocation_type,
+ log_type=self.log_type,
+ client_context=self.client_context,
+ payload=self.payload,
+ qualifier=self.qualifier,
+ )
+ self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
+ if response.get("StatusCode") not in success_status_codes:
+ raise ValueError('Lambda function did not execute', json.dumps(response.get("ResponseMetadata")))
+ if "FunctionError" in response:
+ error_payload_stream = response.get("Payload")
+ error_payload = error_payload_stream.read().decode()
+ raise ValueError(
+ 'Lambda function execution resulted in error',
+ {"ResponseMetadata": response.get("ResponseMetadata"), "Payload": error_payload},
+ )
+ self.log.info('Lambda function invocation succeeded: %r', response.get("ResponseMetadata"))
+ payload_stream = response.get("Payload")
+ payload = payload_stream.read().decode()
+ return payload
Review comment:
```suggestion
payload_stream = response.get("Payload")
payload = payload_stream.read().decode()
if "FunctionError" in response:
raise ValueError(
'Lambda function execution resulted in error',
{"ResponseMetadata": response.get("ResponseMetadata"), "Payload": payload},
)
self.log.info('Lambda function invocation succeeded: %r', response.get("ResponseMetadata"))
return payload
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
potiuk merged pull request #21686:
URL: https://github.com/apache/airflow/pull/21686
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810691604
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
I think the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods if they are not instance-specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] malthe commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
malthe commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810589242
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +40,31 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
+ def invoke_lambda(self, function_name: str, **kwargs):
Review comment:
It would be better if the parameters were spelled out, especially because they now also have typing information. But previously, those parameters also had defaults.
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
Review comment:
This should be `function_name`, etc.
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
Review comment:
This should be refactored a bit and the `kwargs` extraction can be put outside of the `try` handler.
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +40,31 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
+ def invoke_lambda(self, function_name: str, **kwargs):
"""Invoke Lambda Function"""
response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
+ FunctionName=function_name, **{k: v for k, v in kwargs.items() if v is not None}
)
+ return response
+ def create_lambda(
+ self, function_name: str, runtime: str, role: str, handler: str, code: dict, **kwargs
+ ) -> dict:
+ """Create a Lambda Function"""
+ response = self.conn.create_function(
+ FunctionName=function_name,
+ Runtime=runtime,
+ Role=role,
+ Handler=handler,
+ Code=code,
+ **{k: v for k, v in kwargs.items() if v is not None},
Review comment:
What are these `kwargs` – the method is easier to use if they're spelled out to the user.
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
+ for key in self.__dict__
+ if (
+ key
+ in [
+ "Payload",
+ "LogType",
+ "Qualifier",
+ "InvocationType",
+ "ClientContext",
+ ]
+ and self.__dict__[key] is not None
+ )
+ },
+ )
+ self.log.info("Lambda response metadata: %s", json.dumps(response.get("ResponseMetadata")))
Review comment:
I don't think you need to serialize this as JSON; a Python repr string would be fine here:
```python
self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
```
Also, there's a lot of repetition in the error handling 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
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810639416
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
Review comment:
@malthe, I understand. This has been refactored. Kindly review and resolve the conversation if the changes have addressed the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810691604
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
Nitpick: I think the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods if they are not instance-specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815131369
##########
File path: docs/apache-airflow-providers-amazon/operators/lambda.rst
##########
@@ -0,0 +1,54 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+.. _howto/operator:AwsLambdaInvokeFunctionOperator:
+
+AWS Lambda Invoke Function Operator
+===================================
+
+.. contents::
+ :depth: 1
+ :local:
+
+Prerequisite Tasks
+------------------
+
+.. include:: _partials/prerequisite_tasks.rst
+
+Using Operator
+--------------
+Use the
+:class:`~airflow.providers.amazon.aws.operators.aws_lambda.AwsLambdaInvokeFunctionOperator`
+to invoke an AWS Lambda function. To get started with AWS Lambda please visit
+`aws.amazon.com/lambda <https://aws.amazon.com/lambda/>`_
+
+
+In the following example, we invoke an AWS Lambda function.
+
+.. exampleinclude:: /../../airflow/providers/amazon/aws/example_dags/example_lambda.py
+ :language: python
+ :start-after: [START howto_lambda_operator]
+ :end-before: [END howto_lambda_operator]
Review comment:
```suggestion
.. exampleinclude:: /../../airflow/providers/amazon/aws/example_dags/example_lambda.py
:language: python
:dedent: 4
:start-after: [START howto_lambda_operator]
:end-before: [END howto_lambda_operator]
```
Nitpick since you indented the START/END tags.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815145185
##########
File path: docs/apache-airflow-providers-amazon/operators/lambda.rst
##########
@@ -0,0 +1,54 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+.. _howto/operator:AwsLambdaInvokeFunctionOperator:
+
+AWS Lambda Invoke Function Operator
Review comment:
Nitpick. I'm working on standardizing these doc files a bit, if you don't mind.
```suggestion
AWS Lambda Operators
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1051264107
Sorry, I submitted those as single comments. Wish I could retroactively roll them into a review to reduce clutter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1055338827
Awesome work, congrats on your first merged pull request!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810691604
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
Technically, the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods if they are not instance-specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810808306
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
I do not suggest changing this to kwargs, but _keyword-only arguments_.
https://www.python.org/dev/peps/pep-3102/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810639107
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
+ for key in self.__dict__
+ if (
+ key
+ in [
+ "Payload",
+ "LogType",
+ "Qualifier",
+ "InvocationType",
+ "ClientContext",
+ ]
+ and self.__dict__[key] is not None
+ )
+ },
+ )
+ self.log.info("Lambda response metadata: %s", json.dumps(response.get("ResponseMetadata")))
Review comment:
@malthe, thank you for the review. I've made the changes for error handling.
However, I am unable to think of a way to reduce the error handling bit. The errors are handled based on the response from the AWS API. If an exception is not thrown by the underlying boto3 SDK, the handling of the same becomes conditional leading to the multiple `if` blocks.
Would you have any recommendations?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] malthe commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
malthe commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810836179
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
@schirag1993 yeah it's just adding a magic "*" argument to denote "you shall not pass" positional parameters after this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1048429946
Looks like I missed adding documentation and an example DAG. I've updated the PR with the same.
> Note: This needs a major version bump in the Amazon provider.
A few QQs: What does it mean if a major version bump is required? Is there anything required on my part? What usually qualifies for a major version bump for provider packages?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810689118
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
Review comment:
`ZipFile` can be used as a context manager:
```
with zipfile.ZipFile(...) as zip_file:
zip_file.write_str(...)
```
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,103 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.function_name = function_name
+ self.payload = payload
+ self.log_type = log_type
+ self.qualifier = qualifier
+ self.invocation_type = invocation_type
+ self.client_context = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.function_name, self.payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.function_name,
+ invocation_type=self.invocation_type,
+ log_type=self.log_type,
+ client_context=self.client_context,
+ payload=self.payload,
+ qualifier=self.qualifier,
+ )
+ self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
+ except Exception as e:
+ self.log.error(e)
+ raise e
Review comment:
This `except` block doesn't seem needed if you are simply re-printing the error?
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
I think the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods because they are not really instance-specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810691604
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
Nitpick: I think the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods if they are not instance-specific? Although given that the test is using `unittest.TestCase`, maybe it doesn't really matter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1048460234
It means we need to be careful when including this change in a release on PyPI. Nothing you need to do 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
[GitHub] [airflow] schirag1993 commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1053363337
> Apologies for the trouble @potiuk. Working on the fixes today. Let me convert this into a draft.
Fixes pushed! Hope I got it right this time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ferruzzi commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
ferruzzi commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815143113
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,99 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.function_name = function_name
+ self.payload = payload
+ self.log_type = log_type
+ self.qualifier = qualifier
+ self.invocation_type = invocation_type
+ self.client_context = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.function_name, self.payload)
+ response = hook.invoke_lambda(
+ function_name=self.function_name,
+ invocation_type=self.invocation_type,
+ log_type=self.log_type,
+ client_context=self.client_context,
+ payload=self.payload,
+ qualifier=self.qualifier,
+ )
+ self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
+ if response.get("StatusCode") not in success_status_codes:
+ raise ValueError('Lambda function did not execute', json.dumps(response.get("ResponseMetadata")))
+ if "FunctionError" in response:
+ error_payload_stream = response.get("Payload")
+ error_payload = error_payload_stream.read().decode()
+ raise ValueError(
+ 'Lambda function execution resulted in error',
+ {"ResponseMetadata": response.get("ResponseMetadata"), "Payload": error_payload},
+ )
+ self.log.info('Lambda function invocation succeeded: %r', response.get("ResponseMetadata"))
+ payload_stream = response.get("Payload")
+ payload = payload_stream.read().decode()
Review comment:
Seems like Lines 97/98 can be moved before the `if "FunctionError"` block and reused inside the block as well, no? You are getting and decoding the stream whether there is an error or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1052647940
Tests and docs are failing :(
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810755931
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,103 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set `invocation_type` to `Event`. For more details,
+ review the boto3 Lambda invoke docs.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.function_name = function_name
+ self.payload = payload
+ self.log_type = log_type
+ self.qualifier = qualifier
+ self.invocation_type = invocation_type
+ self.client_context = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.function_name, self.payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.function_name,
+ invocation_type=self.invocation_type,
+ log_type=self.log_type,
+ client_context=self.client_context,
+ payload=self.payload,
+ qualifier=self.qualifier,
+ )
+ self.log.info("Lambda response metadata: %r", response.get("ResponseMetadata"))
+ except Exception as e:
+ self.log.error(e)
+ raise e
Review comment:
You're right @edwardwang888, will make 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
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810639439
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
Review comment:
@malthe, done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810639598
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +40,31 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
+ def invoke_lambda(self, function_name: str, **kwargs):
"""Invoke Lambda Function"""
response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
+ FunctionName=function_name, **{k: v for k, v in kwargs.items() if v is not None}
)
+ return response
+ def create_lambda(
+ self, function_name: str, runtime: str, role: str, handler: str, code: dict, **kwargs
+ ) -> dict:
+ """Create a Lambda Function"""
+ response = self.conn.create_function(
+ FunctionName=function_name,
+ Runtime=runtime,
+ Role=role,
+ Handler=handler,
+ Code=code,
+ **{k: v for k, v in kwargs.items() if v is not None},
Review comment:
I understand. The changes have been made @malthe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810639508
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +40,31 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
+ def invoke_lambda(self, function_name: str, **kwargs):
Review comment:
@malthe, I agree. I've made the recommended changes and added defaults.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810691604
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
I think the helper functions (`create_zip`, `create_iam_role`, `create_lambda_function`) can be made static methods if they are not really instance-specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810976375
##########
File path: airflow/providers/amazon/aws/hooks/lambda_function.py
##########
@@ -40,30 +41,84 @@ class LambdaHook(AwsBaseHook):
def __init__(
self,
- function_name: str,
- log_type: str = 'None',
- qualifier: str = '$LATEST',
- invocation_type: str = 'RequestResponse',
*args,
**kwargs,
) -> None:
- self.function_name = function_name
- self.log_type = log_type
- self.invocation_type = invocation_type
- self.qualifier = qualifier
kwargs["client_type"] = "lambda"
super().__init__(*args, **kwargs)
- def invoke_lambda(self, payload: str) -> str:
- """Invoke Lambda Function"""
- response = self.conn.invoke(
- FunctionName=self.function_name,
- InvocationType=self.invocation_type,
- LogType=self.log_type,
- Payload=payload,
- Qualifier=self.qualifier,
- )
+ def invoke_lambda(
+ self,
+ function_name: str,
+ invocation_type: Optional[str] = None,
+ log_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ ):
Review comment:
Ah I misunderstood then. Thank you for the clarifications @uranusjr and @malthe!
Will incorporate the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r810787447
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
Review comment:
Done
##########
File path: tests/providers/amazon/aws/operators/test_lambda.py
##########
@@ -0,0 +1,107 @@
+#
+# 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.
+
+import io
+import json
+import unittest
+import zipfile
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.aws_lambda import AwsLambdaInvokeFunctionOperator
+
+
+@mock_lambda
+@mock_sts
+@mock_iam
+class TestAwsLambdaInvokeFunctionOperator(unittest.TestCase):
+ def test_init(self):
+ lambda_operator = AwsLambdaInvokeFunctionOperator(
+ task_id="test",
+ function_name="test",
+ payload=json.dumps({"TestInput": "Testdata"}),
+ log_type="None",
+ aws_conn_id="aws_conn_test",
+ )
+ assert lambda_operator.task_id == "test"
+ assert lambda_operator.function_name == "test"
+ assert lambda_operator.payload == json.dumps({"TestInput": "Testdata"})
+ assert lambda_operator.log_type == "None"
+ assert lambda_operator.aws_conn_id == "aws_conn_test"
+
+ def create_zip(self, body):
+ code = body
+ zip_output = io.BytesIO()
+ zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
+ zip_file.writestr("lambda_function.py", code)
+ zip_file.close()
+ zip_output.seek(0)
+ return zip_output.read()
+
+ def create_iam_role(self, role_name: str):
+ iam = AwsBaseHook("aws_conn_test", client_type="iam")
+ resp = iam.conn.create_role(
+ RoleName=role_name,
+ AssumeRolePolicyDocument=json.dumps(
+ {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Effect": "Allow",
+ "Principal": {"Service": "lambda.amazonaws.com"},
+ "Action": "sts:AssumeRole",
+ }
+ ],
+ }
+ ),
+ Description="IAM role for Lambda execution.",
+ )
+ return resp["Role"]["Arn"]
+
+ def create_lambda_function(self, function_name: str):
Review comment:
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] schirag1993 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
schirag1993 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r815257438
##########
File path: docs/apache-airflow-providers-amazon/operators/lambda.rst
##########
@@ -0,0 +1,54 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+
+.. _howto/operator:AwsLambdaInvokeFunctionOperator:
+
+AWS Lambda Invoke Function Operator
Review comment:
Not at all. If you have a draft of the docs standardisation (I can view), happy to add anything else I can find.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#discussion_r811614220
##########
File path: airflow/providers/amazon/aws/operators/aws_lambda.py
##########
@@ -0,0 +1,121 @@
+#
+# 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.
+
+import json
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AwsLambdaInvokeFunctionOperator(BaseOperator):
+ """
+ Invokes an AWS Lambda function.
+ You can invoke a function synchronously (and wait for the response),
+ or asynchronously.
+ To invoke a function asynchronously,
+ set InvocationType to `Event`.
+
+ :param function_name: The name of the AWS Lambda function, version, or alias.
+ :param payload: The JSON string that you want to provide to your Lambda function as input.
+ :param log_type: Set to Tail to include the execution log in the response. Otherwise, set to "None".
+ :param qualifier: Specify a version or alias to invoke a published version of the function.
+ :param aws_conn_id: The AWS connection ID to use
+ """
+
+ template_fields: Sequence[str] = ('function_name', 'payload', 'qualifier', 'invocation_type')
+ ui_color = '#ff7300'
+
+ def __init__(
+ self,
+ *,
+ function_name: str,
+ log_type: Optional[str] = None,
+ qualifier: Optional[str] = None,
+ invocation_type: Optional[str] = None,
+ client_context: Optional[str] = None,
+ payload: Optional[str] = None,
+ aws_conn_id='aws_default',
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.FunctionName = function_name
+ self.Payload = payload
+ self.LogType = log_type
+ self.Qualifier = qualifier
+ self.InvocationType = invocation_type
+ self.ClientContext = client_context
+ self.aws_conn_id = aws_conn_id
+
+ def execute(self, context: 'Context'):
+ """
+ Invokes the target AWS Lambda function from Airflow.
+
+ :return: The response payload from the function, or an error object.
+ """
+ hook = LambdaHook(aws_conn_id=self.aws_conn_id)
+ success_status_codes = [200, 202, 204]
+ self.log.info("Invoking AWS Lambda function: %s with payload: %s", self.FunctionName, self.Payload)
+ try:
+ response = hook.invoke_lambda(
+ function_name=self.FunctionName,
+ **{
+ key: self.__dict__[key]
+ for key in self.__dict__
+ if (
+ key
+ in [
+ "Payload",
+ "LogType",
+ "Qualifier",
+ "InvocationType",
+ "ClientContext",
+ ]
+ and self.__dict__[key] is not None
+ )
+ },
+ )
+ self.log.info("Lambda response metadata: %s", json.dumps(response.get("ResponseMetadata")))
Review comment:
@schirag1993 I think @malthe may have been referring to the same thing as my comment, which is already resolved? https://github.com/apache/airflow/pull/21686#discussion_r810686375
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #21686: Feature: Add invoke lambda function operator
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #21686:
URL: https://github.com/apache/airflow/pull/21686#issuecomment-1047472457
The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.
--
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