josh-fell commented on code in PR #28485: URL: https://github.com/apache/airflow/pull/28485#discussion_r1069618377
########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + :param method: request type of the Azure function HTTPTrigger type + :param azure_function_conn_id: The azure function connection ID to use + """ + + conn_name_attr = "azure_functions_conn_id" + default_conn_name = "azure_functions_default" + conn_type = "azure_functions" + hook_name = "Azure Functions" + + def __init__( + self, + method: str = "POST", + azure_function_conn_id: str = default_conn_name, + tcp_keep_alive: bool = True, + tcp_keep_alive_idle: int = 120, + tcp_keep_alive_count: int = 20, + tcp_keep_alive_interval: int = 30, + ) -> None: + super().__init__() + self.azure_function_conn_id = azure_function_conn_id + self.method = method.upper() + self.base_url: str = "" + self.tcp_keep_alive = tcp_keep_alive + self.keep_alive_idle = tcp_keep_alive_idle + self.keep_alive_count = tcp_keep_alive_count + self.keep_alive_interval = tcp_keep_alive_interval + + @staticmethod + def get_connection_form_widgets() -> dict[str, Any]: + """Returns connection widgets to add to connection form""" + from flask_appbuilder.fieldwidgets import BS3TextFieldWidget + from flask_babel import lazy_gettext + from wtforms import StringField + + return { + "tenant_id": StringField( + lazy_gettext("Tenant Id (Active Directory Auth)"), widget=BS3TextFieldWidget() + ) + } + + @staticmethod + def get_ui_field_behaviour() -> dict[str, Any]: + """Returns custom field behaviour""" + return { + "hidden_fields": ["port", "extra"], + "relabeling": { + "host": "Function URL", + "login": "Client Id", + "password": "Client Secret", + "schema": "Scope", + }, + "placeholders": { + "login": "client id", + "password": "client secret", + "host": "https://<APP_NAME>.azurewebsites.net", + "schema": "scope", + "tenant_id": "tenant", + }, Review Comment: ```suggestion "placeholders": {"host": "https://<APP_NAME>.azurewebsites.net"}, ``` IMO placeholders which duplicate the field name aren't _really_ useful. But, toy or pluggable examples are. ########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): Review Comment: It seems wise to inherit from `HttpHook` here. The mechanism built here is already similar or copied from the hook. The [dbt Cloud hook](https://github.com/apache/airflow/blob/main/airflow/providers/dbt/cloud/hooks/dbt.py) should be a very analogous and pertinent example of using `HttpHook`. ########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + :param method: request type of the Azure function HTTPTrigger type + :param azure_function_conn_id: The azure function connection ID to use + """ + + conn_name_attr = "azure_functions_conn_id" + default_conn_name = "azure_functions_default" + conn_type = "azure_functions" + hook_name = "Azure Functions" + + def __init__( + self, + method: str = "POST", + azure_function_conn_id: str = default_conn_name, + tcp_keep_alive: bool = True, + tcp_keep_alive_idle: int = 120, + tcp_keep_alive_count: int = 20, + tcp_keep_alive_interval: int = 30, + ) -> None: + super().__init__() + self.azure_function_conn_id = azure_function_conn_id + self.method = method.upper() + self.base_url: str = "" + self.tcp_keep_alive = tcp_keep_alive + self.keep_alive_idle = tcp_keep_alive_idle + self.keep_alive_count = tcp_keep_alive_count + self.keep_alive_interval = tcp_keep_alive_interval + + @staticmethod + def get_connection_form_widgets() -> dict[str, Any]: + """Returns connection widgets to add to connection form""" + from flask_appbuilder.fieldwidgets import BS3TextFieldWidget + from flask_babel import lazy_gettext + from wtforms import StringField + + return { + "tenant_id": StringField( + lazy_gettext("Tenant Id (Active Directory Auth)"), widget=BS3TextFieldWidget() + ) + } + + @staticmethod + def get_ui_field_behaviour() -> dict[str, Any]: + """Returns custom field behaviour""" + return { + "hidden_fields": ["port", "extra"], + "relabeling": { + "host": "Function URL", + "login": "Client Id", + "password": "Client Secret", + "schema": "Scope", + }, + "placeholders": { + "login": "client id", + "password": "client secret", + "host": "https://<APP_NAME>.azurewebsites.net", + "schema": "scope", + "tenant_id": "tenant", + }, + } + + def get_conn(self, function_key: str | None = None) -> requests.Session: + """ + Returns http session for use with requests + + :param function_key: function key to authenticate + """ + session = requests.Session() + auth_type = "client_key_type" + conn = self.get_connection(self.azure_function_conn_id) + extra = conn.extra_dejson or {} + + if conn.host and "://" in conn.host: + self.base_url = conn.host + + tenant = self._get_field(extra, "tenant_id") + if tenant: + # use Active Directory auth + app_id = conn.login + app_secret = conn.password + scopes = conn.schema + token_credential = ClientSecretCredential(tenant, app_id, app_secret).get_token(scopes).token + elif conn.login and tenant is None: + token_credential = conn.login + auth_type = "functions_key_client" + else: + raise ValueError("Need client id or (tenant, client id, client secret) to authenticate") + headers = self.get_headers(auth_type, token_credential) + session.headers.update(headers) + return session + + def _get_field(self, extra_dict, field_name): Review Comment: This method should not be necessary given that the old extra prefix for connections is [no longer required as of 2.3.0](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html#custom-connection-fields) (the minimum Airflow version for all providers now). The connection doc should notify users to how this connection should be constructed. Also, this is a net-new connection type so there will not be a concern of backward compat or raising an error if users reuse an old connection and upgrade the provider. ########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + :param method: request type of the Azure function HTTPTrigger type + :param azure_function_conn_id: The azure function connection ID to use + """ + + conn_name_attr = "azure_functions_conn_id" + default_conn_name = "azure_functions_default" + conn_type = "azure_functions" + hook_name = "Azure Functions" + + def __init__( + self, + method: str = "POST", + azure_function_conn_id: str = default_conn_name, + tcp_keep_alive: bool = True, + tcp_keep_alive_idle: int = 120, + tcp_keep_alive_count: int = 20, + tcp_keep_alive_interval: int = 30, + ) -> None: + super().__init__() + self.azure_function_conn_id = azure_function_conn_id + self.method = method.upper() + self.base_url: str = "" + self.tcp_keep_alive = tcp_keep_alive + self.keep_alive_idle = tcp_keep_alive_idle + self.keep_alive_count = tcp_keep_alive_count + self.keep_alive_interval = tcp_keep_alive_interval + + @staticmethod + def get_connection_form_widgets() -> dict[str, Any]: + """Returns connection widgets to add to connection form""" + from flask_appbuilder.fieldwidgets import BS3TextFieldWidget + from flask_babel import lazy_gettext + from wtforms import StringField + + return { + "tenant_id": StringField( + lazy_gettext("Tenant Id (Active Directory Auth)"), widget=BS3TextFieldWidget() + ) + } + + @staticmethod + def get_ui_field_behaviour() -> dict[str, Any]: + """Returns custom field behaviour""" + return { + "hidden_fields": ["port", "extra"], + "relabeling": { + "host": "Function URL", + "login": "Client Id", + "password": "Client Secret", + "schema": "Scope", + }, + "placeholders": { + "login": "client id", + "password": "client secret", + "host": "https://<APP_NAME>.azurewebsites.net", + "schema": "scope", + "tenant_id": "tenant", + }, + } + + def get_conn(self, function_key: str | None = None) -> requests.Session: + """ + Returns http session for use with requests + + :param function_key: function key to authenticate + """ + session = requests.Session() + auth_type = "client_key_type" + conn = self.get_connection(self.azure_function_conn_id) + extra = conn.extra_dejson or {} + + if conn.host and "://" in conn.host: + self.base_url = conn.host + + tenant = self._get_field(extra, "tenant_id") + if tenant: + # use Active Directory auth + app_id = conn.login + app_secret = conn.password + scopes = conn.schema + token_credential = ClientSecretCredential(tenant, app_id, app_secret).get_token(scopes).token + elif conn.login and tenant is None: + token_credential = conn.login + auth_type = "functions_key_client" + else: + raise ValueError("Need client id or (tenant, client id, client secret) to authenticate") + headers = self.get_headers(auth_type, token_credential) + session.headers.update(headers) + return session + + def _get_field(self, extra_dict, field_name): + prefix = "extra__wasb__" + if field_name.startswith("extra__"): + raise ValueError( + f"Got prefixed name {field_name}; please remove the '{prefix}' prefix " + f"when using this method." + ) + if field_name in extra_dict: + return extra_dict[field_name] or None + return extra_dict.get(f"{prefix}{field_name}") or None + + @staticmethod + def get_headers(auth_type: str, token_key: str) -> dict[str, Any]: + """Get Headers with auth keys""" + headers: dict[str, Any] = {"Content-Type": "application/json"} + if auth_type == "functions_key_type": + headers["x-functions-key"] = token_key + elif auth_type == "functions_key_client": + headers["x-functions-client"] = token_key + else: + headers["Authorization"] = f"Bearer {token_key}" + return headers + + def invoke_function( + self, + function_name: str, + endpoint: str | None = None, + function_key: str | None = None, + payload: dict[str, Any] | str | None = None, + ) -> Response: + """Invoke Azure Function by making http request with function name and url""" + session = self.get_conn(function_key) + if not endpoint: + endpoint = f"/api/{function_name}" + url = self.url_from_endpoint(endpoint) + if self.method == "GET": + # GET uses params + req = requests.Request(self.method, url, params=payload) + else: + req = requests.Request(self.method, url, data=payload) + prepped_request = session.prepare_request(req) + response = session.send(prepped_request) + + try: + response.raise_for_status() + except requests.exceptions.HTTPError: + self.log.error("HTTP error: %s", response.reason) + self.log.error(response.text) + raise AirflowException(str(response.status_code) + ":" + response.reason) + return response + + def url_from_endpoint(self, endpoint: str | None) -> str: Review Comment: This method can be inherited directly from `HttpHook` instead. ########## docs/apache-airflow-providers-microsoft-azure/connections/azure_function.rst: ########## @@ -0,0 +1,62 @@ + .. 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/connection:azure_functions: + +Microsoft Azure Functions Connection +===================================== + +The Microsoft Azure Functions connection type enables invoking/accessing the functions in the azure functions app. By +using client id, client secret, tenant id and scope get the access token and used than build make a HTTP requests. + +Authenticating to Azure Functions +---------------------------------- + +Currently using token credentials we can able to connect to Azure function using Airflow. + +1. Use `token credentials + <https://docs.microsoft.com/en-us/azure/developer/python/azure-sdk-authenticate?tabs=cmd#authenticate-with-token-credentials>`_ + i.e. add specific credentials (client_id, secret, tenant) and subscription id to the Airflow connection. + +Default Connection IDs +---------------------- + +All hooks and operators related to Microsoft Azure Functions use ``azure_functions_default`` by default. + +Configuring the Connection +-------------------------- + +Function URL + Specify the base URL of the function app. + +Client ID + Specify the ``client_id`` used for the initial connection. + This is needed for *token credentials* authentication mechanism. + +Client Secret + Specify the ``secret`` used for the initial connection. + This is needed for *token credentials* authentication mechanism. + +Tenant ID + Specify the Azure tenant ID used for the initial connection. + This is needed for *token credentials* authentication mechanism. + Use extra param ``tenantId`` to pass in the tenant ID. Review Comment: IMO the connection doc should reference the base field name rather than what's used in the UI since there are more ways to create a connection in Airflow outside the UI than in it. A note about what field each corresponds to in the UI would be helpful though. Again, check out the [dbt Cloud connection doc](https://airflow.apache.org/docs/apache-airflow-providers-dbt-cloud/stable/connections.html#configuring-the-connection). ########## tests/system/providers/microsoft/azure/example_azure_functions.py: ########## @@ -0,0 +1,53 @@ +# 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 json +from datetime import datetime + +from airflow import models +from airflow.providers.microsoft.azure.operators.azure_functions import AzureFunctionsInvokeOperator + +DAG_ID = "example_azure_functions" + +with models.DAG( + DAG_ID, + start_date=datetime(2021, 8, 13), + schedule=None, + catchup=False, + tags=["example"], +) as dag: + + # [START howto_operator_invoke_azure_functions] + invoke_lambda_function = AzureFunctionsInvokeOperator( + task_id="invoke_azure_function", + function_name="test-function-name", + payload=json.dumps({"name": "sample"}), + ) + # [END howto_operator_invoke_azure_functions] + + from tests.system.utils.watcher import watcher + + # This test needs watcher in order to properly mark success/failure + # when "tearDown" task with trigger rule is part of the DAG + list(dag.tasks) >> watcher() Review Comment: ```suggestion ``` I believe this isn't needed since there is only 1 task in the system test. ########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + :param method: request type of the Azure function HTTPTrigger type + :param azure_function_conn_id: The azure function connection ID to use + """ + + conn_name_attr = "azure_functions_conn_id" + default_conn_name = "azure_functions_default" + conn_type = "azure_functions" + hook_name = "Azure Functions" + + def __init__( + self, + method: str = "POST", + azure_function_conn_id: str = default_conn_name, + tcp_keep_alive: bool = True, + tcp_keep_alive_idle: int = 120, + tcp_keep_alive_count: int = 20, + tcp_keep_alive_interval: int = 30, + ) -> None: + super().__init__() + self.azure_function_conn_id = azure_function_conn_id + self.method = method.upper() + self.base_url: str = "" + self.tcp_keep_alive = tcp_keep_alive + self.keep_alive_idle = tcp_keep_alive_idle + self.keep_alive_count = tcp_keep_alive_count + self.keep_alive_interval = tcp_keep_alive_interval + + @staticmethod + def get_connection_form_widgets() -> dict[str, Any]: + """Returns connection widgets to add to connection form""" + from flask_appbuilder.fieldwidgets import BS3TextFieldWidget + from flask_babel import lazy_gettext + from wtforms import StringField + + return { + "tenant_id": StringField( + lazy_gettext("Tenant Id (Active Directory Auth)"), widget=BS3TextFieldWidget() + ) + } + + @staticmethod + def get_ui_field_behaviour() -> dict[str, Any]: + """Returns custom field behaviour""" + return { + "hidden_fields": ["port", "extra"], + "relabeling": { + "host": "Function URL", + "login": "Client Id", + "password": "Client Secret", + "schema": "Scope", + }, + "placeholders": { + "login": "client id", + "password": "client secret", + "host": "https://<APP_NAME>.azurewebsites.net", + "schema": "scope", + "tenant_id": "tenant", + }, + } + + def get_conn(self, function_key: str | None = None) -> requests.Session: + """ + Returns http session for use with requests + + :param function_key: function key to authenticate + """ + session = requests.Session() + auth_type = "client_key_type" + conn = self.get_connection(self.azure_function_conn_id) + extra = conn.extra_dejson or {} + + if conn.host and "://" in conn.host: + self.base_url = conn.host + + tenant = self._get_field(extra, "tenant_id") + if tenant: + # use Active Directory auth + app_id = conn.login + app_secret = conn.password + scopes = conn.schema + token_credential = ClientSecretCredential(tenant, app_id, app_secret).get_token(scopes).token + elif conn.login and tenant is None: + token_credential = conn.login + auth_type = "functions_key_client" + else: + raise ValueError("Need client id or (tenant, client id, client secret) to authenticate") + headers = self.get_headers(auth_type, token_credential) + session.headers.update(headers) + return session + + def _get_field(self, extra_dict, field_name): + prefix = "extra__wasb__" + if field_name.startswith("extra__"): + raise ValueError( + f"Got prefixed name {field_name}; please remove the '{prefix}' prefix " + f"when using this method." + ) + if field_name in extra_dict: + return extra_dict[field_name] or None + return extra_dict.get(f"{prefix}{field_name}") or None + + @staticmethod + def get_headers(auth_type: str, token_key: str) -> dict[str, Any]: + """Get Headers with auth keys""" + headers: dict[str, Any] = {"Content-Type": "application/json"} + if auth_type == "functions_key_type": + headers["x-functions-key"] = token_key + elif auth_type == "functions_key_client": + headers["x-functions-client"] = token_key + else: + headers["Authorization"] = f"Bearer {token_key}" + return headers + + def invoke_function( + self, + function_name: str, + endpoint: str | None = None, + function_key: str | None = None, + payload: dict[str, Any] | str | None = None, Review Comment: WDYT about having the `function_name` and `function_key` being optional parts of the connection and falling back to the connection values if these are not passed in (mainly via the operator calling this method)? This is a similar idea to [handling account ID in dbt Cloud.](https://github.com/apache/airflow/blob/3decb189f786781bb0dfb3420a508a4a2a22bd8b/airflow/providers/dbt/cloud/hooks/dbt.py#L38) (See [connection doc](https://airflow.apache.org/docs/apache-airflow-providers-dbt-cloud/stable/connections.html#configuring-the-connection) too). I have a hunch that most of the time users would be interacting with a single Azure Function for a given connection and instead of adding these values for each operator/task using the connection. This isn't a blocker of course. If you feel differently we can save it as a future enhancement. Just thought it might improve UX a little. ########## airflow/providers/microsoft/azure/provider.yaml: ########## @@ -158,6 +164,9 @@ operators: - integration-name: Microsoft Azure Synapse python-modules: - airflow.providers.microsoft.azure.operators.synapse + - integration-name: Microsoft Azure Functions + python-modules: + - airflow.providers.microsoft.azure.operators.azure_functions Review Comment: To comply with [AIP-21](https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-21%3A+Changes+in+import+paths), the file name shouldn't contain "azure" in the name. Ideally the module path is `airflow.providers.microsoft.azure.operators.functions`. ########## docs/apache-airflow-providers-microsoft-azure/connections/azure_function.rst: ########## @@ -0,0 +1,62 @@ + .. 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/connection:azure_functions: + +Microsoft Azure Functions Connection +===================================== + +The Microsoft Azure Functions connection type enables invoking/accessing the functions in the azure functions app. By +using client id, client secret, tenant id and scope get the access token and used than build make a HTTP requests. + +Authenticating to Azure Functions +---------------------------------- + +Currently using token credentials we can able to connect to Azure function using Airflow. + +1. Use `token credentials + <https://docs.microsoft.com/en-us/azure/developer/python/azure-sdk-authenticate?tabs=cmd#authenticate-with-token-credentials>`_ + i.e. add specific credentials (client_id, secret, tenant) and subscription id to the Airflow connection. + +Default Connection IDs +---------------------- + +All hooks and operators related to Microsoft Azure Functions use ``azure_functions_default`` by default. + +Configuring the Connection +-------------------------- + +Function URL + Specify the base URL of the function app. + +Client ID + Specify the ``client_id`` used for the initial connection. + This is needed for *token credentials* authentication mechanism. + +Client Secret + Specify the ``secret`` used for the initial connection. + This is needed for *token credentials* authentication mechanism. + +Tenant ID + Specify the Azure tenant ID used for the initial connection. + This is needed for *token credentials* authentication mechanism. + Use extra param ``tenantId`` to pass in the tenant ID. Review Comment: ```suggestion Use extra param ``tenant_id`` to pass in the tenant ID. ``` ########## airflow/providers/microsoft/azure/operators/azure_functions.py: ########## @@ -0,0 +1,85 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Sequence + +from airflow.compat.functools import cached_property +from airflow.models import BaseOperator +from airflow.providers.microsoft.azure.hooks.azure_functions import AzureFunctionsHook + +if TYPE_CHECKING: + from airflow.utils.context import Context + + +class AzureFunctionsInvokeOperator(BaseOperator): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + .. seealso:: + For more information on how to use this operator, take a look at the guide: + :ref:`howto/operator:AzureFunctionsInvokeOperator` + + :param function_name: The name of the Azure function. + :param function_key: function level auth key. + :param endpoint_url: endpoint url. + :param method_type: request type of the Azure function HTTPTrigger type + :param payload: JSON provided as input to the azure function + :param azure_function_conn_id: The azure function connection ID to use + """ + + template_fields: Sequence[str] = ("function_name", "payload") + ui_color = "#ff7300" + + def __init__( + self, + *, + function_name: str, + function_key: str | None = None, + endpoint_url: str | None = None, + method_type: str = "POST", + payload: dict[str, Any] | str | None = None, + azure_function_conn_id: str = "azure_functions_default", + **kwargs, + ): + super().__init__(**kwargs) + self.function_name = function_name + self.function_key = function_key + self.payload = payload + self.method_type = method_type + self.endpoint_url = endpoint_url + self.azure_function_conn_id = azure_function_conn_id + + @cached_property + def hook(self) -> AzureFunctionsHook: + return AzureFunctionsHook(azure_function_conn_id=self.azure_function_conn_id, method=self.method_type) Review Comment: It doesn't seem like this needs to be a cached property. The hook object only used in the `execute()` method and probably could be instantiated there. ########## airflow/providers/microsoft/azure/provider.yaml: ########## @@ -209,6 +218,9 @@ hooks: - integration-name: Microsoft Azure Synapse python-modules: - airflow.providers.microsoft.azure.hooks.synapse + - integration-name: Microsoft Azure Functions + python-modules: + - airflow.providers.microsoft.azure.hooks.azure_functions Review Comment: Same AIP-21 comment here. ########## docs/apache-airflow-providers-microsoft-azure/connections/azure_function.rst: ########## @@ -0,0 +1,62 @@ + .. 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/connection:azure_functions: + +Microsoft Azure Functions Connection +===================================== + +The Microsoft Azure Functions connection type enables invoking/accessing the functions in the azure functions app. By Review Comment: ```suggestion The Microsoft Azure Functions connection type enables invoking/accessing the functions in the Azure functions app. By ``` A nit but for consistency in the doc. ########## airflow/providers/microsoft/azure/hooks/azure_functions.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from typing import Any + +import requests +from azure.identity import ClientSecretCredential +from requests import Response + +from airflow.exceptions import AirflowException +from airflow.hooks.base import BaseHook + + +class AzureFunctionsHook(BaseHook): + """ + Invokes an Azure function. You can invoke a function in azure by making http request + + :param method: request type of the Azure function HTTPTrigger type + :param azure_function_conn_id: The azure function connection ID to use + """ + + conn_name_attr = "azure_functions_conn_id" + default_conn_name = "azure_functions_default" + conn_type = "azure_functions" + hook_name = "Azure Functions" + + def __init__( + self, + method: str = "POST", + azure_function_conn_id: str = default_conn_name, + tcp_keep_alive: bool = True, + tcp_keep_alive_idle: int = 120, + tcp_keep_alive_count: int = 20, + tcp_keep_alive_interval: int = 30, + ) -> None: + super().__init__() + self.azure_function_conn_id = azure_function_conn_id + self.method = method.upper() + self.base_url: str = "" + self.tcp_keep_alive = tcp_keep_alive + self.keep_alive_idle = tcp_keep_alive_idle + self.keep_alive_count = tcp_keep_alive_count + self.keep_alive_interval = tcp_keep_alive_interval + + @staticmethod + def get_connection_form_widgets() -> dict[str, Any]: + """Returns connection widgets to add to connection form""" + from flask_appbuilder.fieldwidgets import BS3TextFieldWidget + from flask_babel import lazy_gettext + from wtforms import StringField + + return { + "tenant_id": StringField( + lazy_gettext("Tenant Id (Active Directory Auth)"), widget=BS3TextFieldWidget() + ) + } + + @staticmethod + def get_ui_field_behaviour() -> dict[str, Any]: + """Returns custom field behaviour""" + return { + "hidden_fields": ["port", "extra"], + "relabeling": { + "host": "Function URL", + "login": "Client Id", + "password": "Client Secret", + "schema": "Scope", + }, + "placeholders": { + "login": "client id", + "password": "client secret", + "host": "https://<APP_NAME>.azurewebsites.net", + "schema": "scope", + "tenant_id": "tenant", + }, Review Comment: An example of what the `Scope` value could be seems useful here actually. ########## docs/apache-airflow-providers-microsoft-azure/operators/azure_functions.rst: ########## @@ -0,0 +1,47 @@ + .. Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + .. http://www.apache.org/licenses/LICENSE-2.0 + + .. Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + +Azure Functions Operators +========================= +Azure Functions is a serverless solution that allows you to write less code, maintain less infrastructure, +and save on costs. Instead of worrying about deploying and maintaining servers, the cloud infrastructure provides +all the up-to-date resources needed to keep your applications running. Azure Functions allows you to implement your +system's logic into readily available blocks of code. These code blocks are called "functions". + +Operators +--------- + +.. _howto/operator:AzureFunctionsInvokeOperator: + +Invoke an Azure functions +------------------------- +Use the :class:`~airflow.providers.microsoft.azure.operators.azure_functions.AzureFunctionsInvokeOperator` to invoke the +functions in azure. Below is an example of using this operator. + + .. exampleinclude:: /../../tests/system/providers/microsoft/azure/example_azure_functions.py + :language: python + :dedent: 0 Review Comment: ```suggestion :dedent: 4 ``` Since the `START` marker is indented in the example DAG file. Otherwise there will be some left hand whitespace/indent in the displayed code snippet. ########## docs/apache-airflow-providers-microsoft-azure/connections/azure_function.rst: ########## @@ -0,0 +1,62 @@ + .. 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/connection:azure_functions: + +Microsoft Azure Functions Connection Review Comment: It would great to see examples of how to build the connection via a URI in this doc too. -- 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]
