jroachgolf84 commented on code in PR #74616: URL: https://github.com/apache/airflow/pull/74616#discussion_r4502753501
########## providers/microsoft/azure/docs/operators/container_apps.rst: ########## @@ -0,0 +1,90 @@ + .. 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. + +Azure Container Apps Jobs +========================= + +.. _howto/operator:AzureContainerAppsJobOperator: + +Use :class:`~airflow.providers.microsoft.azure.operators.container_apps.AzureContainerAppsJobOperator` +to start an **existing** Azure Container Apps Job and wait for the execution returned by Azure. +The operator does not create, update, or delete jobs. On success, its return value (and default +``return_value`` XCom) is the execution name. Review Comment: Does it make sense to also create an Operator that creates jobs? I know that we do that quite a bit in other providers. ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_apps.py: ########## Review Comment: Stylistically, I'd recommend adding some spaces to your code, everything is a bit bunched up. ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_apps.py: ########## @@ -0,0 +1,212 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +from functools import cached_property +from typing import TYPE_CHECKING, Any + +from azure.core.exceptions import AzureError, ServiceRequestTimeoutError, ServiceResponseTimeoutError + +from airflow.providers.common.compat.sdk import BaseOperator, conf +from airflow.providers.microsoft.azure.exceptions import AzureContainerAppsJobExecutionError +from airflow.providers.microsoft.azure.hooks.container_apps import ( + AzureContainerAppsJobHook, + get_execution_result, +) +from airflow.providers.microsoft.azure.triggers.container_apps import AzureContainerAppsJobTrigger + +if TYPE_CHECKING: + from airflow.providers.common.compat.sdk import Context + + +class AzureContainerAppsJobOperator(BaseOperator): + """ + Start an existing Azure Container Apps Job and wait for its exact execution. + + Set task ``retries=0``: after an ambiguous submission response, retrying may + start another execution. Azure transport retries are disabled for submission. + + :param resource_group: Resource group containing the job. + :param job_name: Name of the existing job. + :param azure_conn_id: Azure connection ID. + :param subscription_id: Subscription ID; defaults to connection extra ``subscriptionId``. + :param container_name: Regular container to override. Required with any overrides. + :param image: Replacement image for the named container. + :param command: Replacement command for the named container. + :param args: Replacement arguments for the named container. + :param env_vars: Replacement environment for the named container, not a merge. + :param cpu: Replacement CPU allocation for the named container. + :param memory: Replacement memory allocation, for example ``"1Gi"``. + :param deferrable: Release the worker while waiting for the execution. + :param poll_interval_seconds: Seconds between execution status polls. + :param timeout_seconds: Maximum wait after submission, or None for no deadline. + :param cancel_on_termination: Stop this execution on explicit termination. + Deferred cancellation requires Airflow support for trigger ``on_kill``. + """ + + template_fields: Sequence[str] = ( + "azure_conn_id", + "subscription_id", + "resource_group", + "job_name", + "container_name", + "image", + "command", + "args", + "env_vars", + "cpu", + "memory", + ) + template_fields_renderers = {"command": "json", "args": "json", "env_vars": "json"} + + def __init__( + self, + *, + resource_group: str, + job_name: str, + azure_conn_id: str = "azure_default", + subscription_id: str | None = None, + container_name: str | None = None, + image: str | None = None, + command: Sequence[str] | None = None, + args: Sequence[str] | None = None, + env_vars: Mapping[str, str] | None = None, + cpu: float | str | None = None, + memory: str | None = None, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + poll_interval_seconds: float = 60, + timeout_seconds: float | None = None, + cancel_on_termination: bool = True, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + if poll_interval_seconds <= 0: + raise ValueError("poll_interval_seconds must be greater than zero") + if timeout_seconds is not None and timeout_seconds <= 0: + raise ValueError("timeout_seconds must be greater than zero") Review Comment: ```suggestion raise ValueError("timeout_seconds must be greater than zero") ``` ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_apps.py: ########## @@ -0,0 +1,212 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +from functools import cached_property +from typing import TYPE_CHECKING, Any + +from azure.core.exceptions import AzureError, ServiceRequestTimeoutError, ServiceResponseTimeoutError + +from airflow.providers.common.compat.sdk import BaseOperator, conf +from airflow.providers.microsoft.azure.exceptions import AzureContainerAppsJobExecutionError +from airflow.providers.microsoft.azure.hooks.container_apps import ( + AzureContainerAppsJobHook, + get_execution_result, +) +from airflow.providers.microsoft.azure.triggers.container_apps import AzureContainerAppsJobTrigger + +if TYPE_CHECKING: + from airflow.providers.common.compat.sdk import Context + + +class AzureContainerAppsJobOperator(BaseOperator): + """ + Start an existing Azure Container Apps Job and wait for its exact execution. + + Set task ``retries=0``: after an ambiguous submission response, retrying may + start another execution. Azure transport retries are disabled for submission. + + :param resource_group: Resource group containing the job. + :param job_name: Name of the existing job. + :param azure_conn_id: Azure connection ID. + :param subscription_id: Subscription ID; defaults to connection extra ``subscriptionId``. + :param container_name: Regular container to override. Required with any overrides. + :param image: Replacement image for the named container. + :param command: Replacement command for the named container. + :param args: Replacement arguments for the named container. + :param env_vars: Replacement environment for the named container, not a merge. + :param cpu: Replacement CPU allocation for the named container. + :param memory: Replacement memory allocation, for example ``"1Gi"``. + :param deferrable: Release the worker while waiting for the execution. + :param poll_interval_seconds: Seconds between execution status polls. + :param timeout_seconds: Maximum wait after submission, or None for no deadline. + :param cancel_on_termination: Stop this execution on explicit termination. + Deferred cancellation requires Airflow support for trigger ``on_kill``. + """ + + template_fields: Sequence[str] = ( + "azure_conn_id", + "subscription_id", + "resource_group", + "job_name", + "container_name", + "image", + "command", + "args", + "env_vars", + "cpu", + "memory", + ) + template_fields_renderers = {"command": "json", "args": "json", "env_vars": "json"} + + def __init__( + self, + *, + resource_group: str, + job_name: str, + azure_conn_id: str = "azure_default", + subscription_id: str | None = None, + container_name: str | None = None, + image: str | None = None, + command: Sequence[str] | None = None, + args: Sequence[str] | None = None, + env_vars: Mapping[str, str] | None = None, + cpu: float | str | None = None, + memory: str | None = None, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + poll_interval_seconds: float = 60, + timeout_seconds: float | None = None, + cancel_on_termination: bool = True, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + if poll_interval_seconds <= 0: + raise ValueError("poll_interval_seconds must be greater than zero") + if timeout_seconds is not None and timeout_seconds <= 0: + raise ValueError("timeout_seconds must be greater than zero") + self.resource_group = resource_group + self.job_name = job_name + self.azure_conn_id = azure_conn_id + self.subscription_id = subscription_id + self.container_name = container_name + self.image = image + self.command = command + self.args = args + self.env_vars = env_vars + self.cpu = cpu + self.memory = memory + self.deferrable = deferrable + self.poll_interval_seconds = poll_interval_seconds + self.timeout_seconds = timeout_seconds + self.cancel_on_termination = cancel_on_termination + self._execution_name: str | None = None + self._terminal_observed = False + + @cached_property + def hook(self) -> AzureContainerAppsJobHook: + """Create the hook after target and override fields have been templated.""" + return AzureContainerAppsJobHook(self.azure_conn_id, self.subscription_id) + + def execute(self, context: Context) -> str: + """Submit once, then poll on the worker or defer to the triggerer.""" + if not self.resource_group or not self.job_name: Review Comment: I haven't seen this sort of validation in other Microsoft Azure Operators/Triggers/Sensors, etc. Are you sure that `resource_group_name` is required? ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_apps.py: ########## @@ -0,0 +1,212 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +from functools import cached_property +from typing import TYPE_CHECKING, Any + +from azure.core.exceptions import AzureError, ServiceRequestTimeoutError, ServiceResponseTimeoutError + +from airflow.providers.common.compat.sdk import BaseOperator, conf +from airflow.providers.microsoft.azure.exceptions import AzureContainerAppsJobExecutionError +from airflow.providers.microsoft.azure.hooks.container_apps import ( + AzureContainerAppsJobHook, + get_execution_result, +) +from airflow.providers.microsoft.azure.triggers.container_apps import AzureContainerAppsJobTrigger + +if TYPE_CHECKING: + from airflow.providers.common.compat.sdk import Context + + +class AzureContainerAppsJobOperator(BaseOperator): + """ + Start an existing Azure Container Apps Job and wait for its exact execution. + + Set task ``retries=0``: after an ambiguous submission response, retrying may + start another execution. Azure transport retries are disabled for submission. + + :param resource_group: Resource group containing the job. + :param job_name: Name of the existing job. + :param azure_conn_id: Azure connection ID. + :param subscription_id: Subscription ID; defaults to connection extra ``subscriptionId``. + :param container_name: Regular container to override. Required with any overrides. + :param image: Replacement image for the named container. + :param command: Replacement command for the named container. + :param args: Replacement arguments for the named container. + :param env_vars: Replacement environment for the named container, not a merge. + :param cpu: Replacement CPU allocation for the named container. + :param memory: Replacement memory allocation, for example ``"1Gi"``. + :param deferrable: Release the worker while waiting for the execution. + :param poll_interval_seconds: Seconds between execution status polls. + :param timeout_seconds: Maximum wait after submission, or None for no deadline. + :param cancel_on_termination: Stop this execution on explicit termination. + Deferred cancellation requires Airflow support for trigger ``on_kill``. + """ + + template_fields: Sequence[str] = ( + "azure_conn_id", + "subscription_id", + "resource_group", + "job_name", + "container_name", + "image", + "command", + "args", + "env_vars", + "cpu", + "memory", + ) + template_fields_renderers = {"command": "json", "args": "json", "env_vars": "json"} + + def __init__( + self, + *, + resource_group: str, + job_name: str, + azure_conn_id: str = "azure_default", + subscription_id: str | None = None, + container_name: str | None = None, + image: str | None = None, + command: Sequence[str] | None = None, + args: Sequence[str] | None = None, + env_vars: Mapping[str, str] | None = None, + cpu: float | str | None = None, + memory: str | None = None, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + poll_interval_seconds: float = 60, + timeout_seconds: float | None = None, + cancel_on_termination: bool = True, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) Review Comment: ```suggestion super().__init__(**kwargs) ``` ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_apps.py: ########## @@ -0,0 +1,212 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +from functools import cached_property +from typing import TYPE_CHECKING, Any + +from azure.core.exceptions import AzureError, ServiceRequestTimeoutError, ServiceResponseTimeoutError + +from airflow.providers.common.compat.sdk import BaseOperator, conf +from airflow.providers.microsoft.azure.exceptions import AzureContainerAppsJobExecutionError +from airflow.providers.microsoft.azure.hooks.container_apps import ( + AzureContainerAppsJobHook, + get_execution_result, +) +from airflow.providers.microsoft.azure.triggers.container_apps import AzureContainerAppsJobTrigger + +if TYPE_CHECKING: + from airflow.providers.common.compat.sdk import Context + + +class AzureContainerAppsJobOperator(BaseOperator): + """ + Start an existing Azure Container Apps Job and wait for its exact execution. + + Set task ``retries=0``: after an ambiguous submission response, retrying may + start another execution. Azure transport retries are disabled for submission. + + :param resource_group: Resource group containing the job. + :param job_name: Name of the existing job. + :param azure_conn_id: Azure connection ID. + :param subscription_id: Subscription ID; defaults to connection extra ``subscriptionId``. + :param container_name: Regular container to override. Required with any overrides. + :param image: Replacement image for the named container. + :param command: Replacement command for the named container. + :param args: Replacement arguments for the named container. + :param env_vars: Replacement environment for the named container, not a merge. + :param cpu: Replacement CPU allocation for the named container. + :param memory: Replacement memory allocation, for example ``"1Gi"``. + :param deferrable: Release the worker while waiting for the execution. + :param poll_interval_seconds: Seconds between execution status polls. + :param timeout_seconds: Maximum wait after submission, or None for no deadline. + :param cancel_on_termination: Stop this execution on explicit termination. + Deferred cancellation requires Airflow support for trigger ``on_kill``. + """ + + template_fields: Sequence[str] = ( + "azure_conn_id", + "subscription_id", + "resource_group", + "job_name", + "container_name", + "image", + "command", + "args", + "env_vars", + "cpu", + "memory", + ) + template_fields_renderers = {"command": "json", "args": "json", "env_vars": "json"} + + def __init__( + self, + *, + resource_group: str, + job_name: str, + azure_conn_id: str = "azure_default", + subscription_id: str | None = None, + container_name: str | None = None, + image: str | None = None, + command: Sequence[str] | None = None, + args: Sequence[str] | None = None, + env_vars: Mapping[str, str] | None = None, + cpu: float | str | None = None, + memory: str | None = None, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + poll_interval_seconds: float = 60, + timeout_seconds: float | None = None, + cancel_on_termination: bool = True, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + if poll_interval_seconds <= 0: + raise ValueError("poll_interval_seconds must be greater than zero") Review Comment: ```suggestion raise ValueError("poll_interval_seconds must be greater than zero") ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
