This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new dbbc8163108 Support unexposed CreateWorkflow and UpdateWorkflow inputs
in MWAA Serverless operators (#73408)
dbbc8163108 is described below
commit dbbc8163108ec7840676ecfce90a7c29c498e100
Author: Kazushi Y <[email protected]>
AuthorDate: Fri Sep 25 02:42:48 2026 +0900
Support unexposed CreateWorkflow and UpdateWorkflow inputs in MWAA
Serverless operators (#73408)
MwaaServerlessCreateWorkflowOperator reaches 6 of the 12 CreateWorkflow
request
parameters and MwaaServerlessUpdateWorkflowOperator 5 of the 9 for
UpdateWorkflow.
Neither accepts a kwargs passthrough, so NetworkConfiguration,
LoggingConfiguration, EngineVersion and TriggerMode are unreachable, along
with
EncryptionConfiguration and ClientToken on create.
Add create_workflow_kwargs and update_workflow_kwargs, following the
passthrough
convention the provider already uses for CreateCluster and
CreateApplication.
Each dict is splatted alongside the explicit parameters, so a key that
collides
with one the operator already sets raises rather than silently overriding
it.
---
.../amazon/aws/operators/mwaa_serverless.py | 41 +++++++++++--
.../amazon/aws/operators/test_mwaa_serverless.py | 70 ++++++++++++++++++++++
2 files changed, 105 insertions(+), 6 deletions(-)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/mwaa_serverless.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/mwaa_serverless.py
index 142affd82fe..d0f8883c90d 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/operators/mwaa_serverless.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/operators/mwaa_serverless.py
@@ -104,15 +104,28 @@ class
MwaaServerlessCreateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
:param role_arn: The execution role ARN. (templated)
:param description: Optional description. (templated)
:param tags: Optional tags dict.
+ :param create_workflow_kwargs: Optional parameters to pass to the
``CreateWorkflow`` API for
+ inputs this operator does not expose, such as ``NetworkConfiguration``,
+ ``LoggingConfiguration``, ``EngineVersion``, ``TriggerMode``,
``EncryptionConfiguration``
+ and ``ClientToken``. (templated)
:param if_exists: Behavior when the workflow already exists.
``"fail"`` raises an error, ``"skip"`` returns the existing ARN.
"""
aws_hook_class = AwsBaseHook
template_fields: tuple[str, ...] = aws_template_fields(
- "workflow_name", "definition_s3_location", "code", "role_arn",
"description"
+ "workflow_name",
+ "definition_s3_location",
+ "code",
+ "role_arn",
+ "description",
+ "create_workflow_kwargs",
)
- template_fields_renderers = {"definition_s3_location": "json", "code":
"json"}
+ template_fields_renderers = {
+ "definition_s3_location": "json",
+ "code": "json",
+ "create_workflow_kwargs": "json",
+ }
def __init__(
self,
@@ -123,6 +136,7 @@ class
MwaaServerlessCreateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
role_arn: str,
description: str | None = None,
tags: dict[str, str] | None = None,
+ create_workflow_kwargs: dict[str, Any] | None = None,
if_exists: Literal["fail", "skip"] = "skip",
**kwargs,
) -> None:
@@ -133,6 +147,7 @@ class
MwaaServerlessCreateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
self.role_arn = role_arn
self.description = description
self.tags = tags
+ self.create_workflow_kwargs = create_workflow_kwargs or {}
self.if_exists = if_exists
@property
@@ -152,7 +167,7 @@ class
MwaaServerlessCreateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
}
)
try:
- response = self.hook.conn.create_workflow(**kwargs)
+ response = self.hook.conn.create_workflow(**kwargs,
**self.create_workflow_kwargs)
workflow_arn = response["WorkflowArn"]
except ClientError as e:
if e.response["Error"]["Code"] == "ConflictException" and
self.if_exists == "skip":
@@ -185,13 +200,25 @@ class
MwaaServerlessUpdateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
optionally ``VersionId``. (templated)
:param role_arn: The execution role ARN. (templated)
:param description: Optional updated description. (templated)
+ :param update_workflow_kwargs: Optional parameters to pass to the
``UpdateWorkflow`` API for
+ inputs this operator does not expose, such as ``NetworkConfiguration``,
+ ``LoggingConfiguration``, ``EngineVersion`` and ``TriggerMode``.
(templated)
"""
aws_hook_class = AwsBaseHook
template_fields: tuple[str, ...] = aws_template_fields(
- "workflow_arn", "definition_s3_location", "code", "role_arn",
"description"
+ "workflow_arn",
+ "definition_s3_location",
+ "code",
+ "role_arn",
+ "description",
+ "update_workflow_kwargs",
)
- template_fields_renderers = {"definition_s3_location": "json", "code":
"json"}
+ template_fields_renderers = {
+ "definition_s3_location": "json",
+ "code": "json",
+ "update_workflow_kwargs": "json",
+ }
def __init__(
self,
@@ -201,6 +228,7 @@ class
MwaaServerlessUpdateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
code: dict[str, Any] | None = None,
role_arn: str,
description: str | None = None,
+ update_workflow_kwargs: dict[str, Any] | None = None,
**kwargs,
) -> None:
super().__init__(**kwargs)
@@ -209,6 +237,7 @@ class
MwaaServerlessUpdateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
self.code = code
self.role_arn = role_arn
self.description = description
+ self.update_workflow_kwargs = update_workflow_kwargs or {}
@property
def _hook_parameters(self) -> dict[str, Any]:
@@ -225,7 +254,7 @@ class
MwaaServerlessUpdateWorkflowOperator(AwsBaseOperator[AwsBaseHook]):
"Description": self.description,
}
)
- response = self.hook.conn.update_workflow(**kwargs)
+ response = self.hook.conn.update_workflow(**kwargs,
**self.update_workflow_kwargs)
workflow_arn = response["WorkflowArn"]
self.log.info("Workflow %s updated to version %s", workflow_arn,
response.get("WorkflowVersion"))
return workflow_arn
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_mwaa_serverless.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_mwaa_serverless.py
index 3397120f70c..6937deded0c 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_mwaa_serverless.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_mwaa_serverless.py
@@ -90,6 +90,26 @@ WORKFLOW_ARN =
"arn:aws:mwaa-serverless:us-east-1:123456789012:workflow/test-wor
S3_LOCATION = {"Bucket": "test-bucket", "ObjectKey": "workflow.yaml"}
CODE = {"S3Location": {"Bucket": "test-bucket", "ObjectKey":
"code/my_package.zip"}}
ROLE_ARN = "arn:aws:iam::123456789012:role/test-role"
+CREATE_WORKFLOW_KWARGS = {
+ "NetworkConfiguration": {
+ "SubnetIds": ["subnet-0123456789abcdef0", "subnet-0fedcba9876543210"],
+ "SecurityGroupIds": ["sg-0123456789abcdef0"],
+ },
+ "LoggingConfiguration": {"LogGroupName":
"/aws/mwaa-serverless/test-workflow"},
+ "EncryptionConfiguration": {"Type": "AWS_MANAGED_KEY"},
+ "EngineVersion": 1,
+ "TriggerMode": "manual_only",
+ "ClientToken": "test-client-token",
+}
+UPDATE_WORKFLOW_KWARGS = {
+ "NetworkConfiguration": {
+ "SubnetIds": ["subnet-0123456789abcdef0", "subnet-0fedcba9876543210"],
+ "SecurityGroupIds": ["sg-0123456789abcdef0"],
+ },
+ "LoggingConfiguration": {"LogGroupName":
"/aws/mwaa-serverless/test-workflow"},
+ "EngineVersion": 1,
+ "TriggerMode": "manual_only",
+}
class TestMwaaServerlessCreateWorkflowOperator:
@@ -178,6 +198,29 @@ class TestMwaaServerlessCreateWorkflowOperator:
)
assert result == WORKFLOW_ARN
+ @mock.patch.object(AwsBaseHook, "conn", new_callable=mock.PropertyMock)
+ def test_execute_with_create_workflow_kwargs(self, mock_conn):
+ op = MwaaServerlessCreateWorkflowOperator(
+ task_id="create_workflow",
+ workflow_name=WORKFLOW_NAME,
+ definition_s3_location=S3_LOCATION,
+ role_arn=ROLE_ARN,
+ create_workflow_kwargs=CREATE_WORKFLOW_KWARGS,
+ )
+ mock_client = mock.MagicMock()
+ mock_client.create_workflow.return_value = {"WorkflowArn":
WORKFLOW_ARN}
+ mock_conn.return_value = mock_client
+
+ result = op.execute({})
+
+ mock_client.create_workflow.assert_called_once_with(
+ Name=WORKFLOW_NAME,
+ DefinitionS3Location=S3_LOCATION,
+ RoleArn=ROLE_ARN,
+ **CREATE_WORKFLOW_KWARGS,
+ )
+ assert result == WORKFLOW_ARN
+
def test_template_fields(self):
validate_template_fields(self.operator)
@@ -275,6 +318,33 @@ class TestMwaaServerlessUpdateWorkflowOperator:
)
assert result == WORKFLOW_ARN
+ @mock.patch.object(AwsBaseHook, "conn", new_callable=mock.PropertyMock)
+ def test_execute_with_update_workflow_kwargs(self, mock_conn):
+ op = MwaaServerlessUpdateWorkflowOperator(
+ task_id="update_workflow",
+ workflow_arn=WORKFLOW_ARN,
+ definition_s3_location=S3_LOCATION,
+ role_arn=ROLE_ARN,
+ update_workflow_kwargs=UPDATE_WORKFLOW_KWARGS,
+ )
+ mock_client = mock.MagicMock()
+ mock_client.update_workflow.return_value = {
+ "WorkflowArn": WORKFLOW_ARN,
+ "WorkflowVersion": "abc123",
+ "ModifiedAt": "2026-05-12T00:00:00Z",
+ }
+ mock_conn.return_value = mock_client
+
+ result = op.execute({})
+
+ mock_client.update_workflow.assert_called_once_with(
+ WorkflowArn=WORKFLOW_ARN,
+ DefinitionS3Location=S3_LOCATION,
+ RoleArn=ROLE_ARN,
+ **UPDATE_WORKFLOW_KWARGS,
+ )
+ assert result == WORKFLOW_ARN
+
def test_template_fields(self):
validate_template_fields(self.operator)
