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 2020/09/18 14:59:51 UTC
[GitHub] [airflow] flvndh opened a new pull request #11015: Add Azure Data Factory hook
flvndh opened a new pull request #11015:
URL: https://github.com/apache/airflow/pull/11015
Tackle #10995
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-769864788
[The Workflow run](https://github.com/apache/airflow/actions/runs/521153013) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r510711932
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
If both does the same shouldn't we only have one method, too? like `create_or_update`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-720995737
LGTM. Can you resolve the failing static checks, please? :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-786128643
The PR was some 200 commits behind of Master -- so rebased and pushed --let's wait for the CI results again
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-777643200
> > What is the status of this PR? How can others help get this across the line?
>
> @don4of4 I consider it as being ready. I'm patiently waiting for someone to merge it or tell me what is blocking 🤷♂️
Pylint errors needs to be fixed just liked @feluelle suggested:
https://github.com/apache/airflow/pull/11015#issuecomment-770997205
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-753691650
@flvndh Can you please resolve the conflicts and ping us again, thanks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514943917
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
Nice! I didn't know `raise from`. Looks really good IMO.
If that also is to generic you can also check for a kind of code or message?!
Sth like.
```python
except CloudError as e:
if e.status_or_code_or_message == 1234:
raise AirflowException(f"Factory {factory_name!r} already exists") from e
raise
```
WDYT?
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
Nice! I didn't know `raise from`. Looks really good IMO.
If that is too generic you can also check for a kind of code or message?!
Sth like.
```python
except CloudError as e:
if e.status_or_code_or_message == 1234:
raise AirflowException(f"Factory {factory_name!r} already exists") from e
raise
```
WDYT?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-715486999
[The Workflow run](https://github.com/apache/airflow/actions/runs/324648034) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514955151
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I'm glad you learned something from me 😋
Unfortunately, when you try to access something that doesn't exist it fails with a "403 - Unauthorized":
```
The client 'xxxx' with object id 'xxxx' does not have authorization to perform action 'Microsoft.DataFactory/factories/read' over scope '/subscriptions/xxxx/resourceGroups/xxxx/providers/Microsoft.DataFactory/factories/xxxx' or the scope is invalid. If access was recently granted, please refresh your credentials.
```
This could also be the case if the user doesn't set his credentials correctly. 😕
[Edit]
Thinking a bit more about it, if there is something wrong with the user authorizations the second call would anyway fail with the same error so this is not misleading the user. Thus, problem solved for the `create` method but the same applies if we want to check that the factory exists when we call `update`.
We could get around this using `factories.list_by_resource_group()` ([doc](https://docs.microsoft.com/en-us/python/api/azure-mgmt-datafactory/azure.mgmt.datafactory.operations.factoriesoperations?view=azure-python#list-by-resource-group-resource-group-name--custom-headers-none--raw-false----operation-config-)) instead of the `factories.get` method 😃
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r501609552
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection during `__init__` because init will be called every time the dag file will be parsed - more than it actually needs. You only want it to be called when an actual task runs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514452717
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I had something like this in mind for the checks:
```python
@provide_targeted_factory
def create_factory(
self,
factory: Factory,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> Factory:
"""
Create the factory.
:param factory: The factory resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:return: The factory.
"""
factories = self.get_conn().factories
try:
factories.get(resource_group_name, factory_name)
except CloudError as e:
raise AirflowException(f"Factory {factory_name!r} already exists") from e
else:
return factories.create_or_update(resource_group_name, factory_name, factory, **config)
```
but this `CloudError` is a little too generic to my taste. What would you do ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-718949985
[The Workflow run](https://github.com/apache/airflow/actions/runs/336480290) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-777639763
> What is the status of this PR? How can others help get this across the line?
@don4of4 I consider it as being ready. I'm patiently waiting for someone to merge it or tell me what is blocking 🤷♂️
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r501621234
##########
File path: tests/providers/microsoft/azure/hooks/test_azure_data_factory.py
##########
@@ -0,0 +1,440 @@
+# 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 re import A
+from unittest.mock import patch
+
+import pytest
+from pytest import fixture
+
+from airflow.exceptions import AirflowException
+from airflow.models.connection import Connection
+from airflow.providers.microsoft.azure.hooks.azure_data_factory import AzureDataFactoryHook
+from airflow.utils import db
+
+
+DEFAULT_RESOURCE_GROUP = "defaultResourceGroup"
+RESOURCE_GROUP = "testResourceGroup"
+
+DEFAULT_FACTORY = "defaultFactory"
+FACTORY = "testFactory"
+
+MODEL = object()
+
+
+def setup_module(module):
+ connection = Connection(
+ conn_id="azure_data_factory_test",
+ conn_type="azure_data_factory",
+ login="clientId",
+ password="clientSecret",
+ extra=json.dumps(
+ {
+ "tenantId": "tenantId",
+ "subscriptionId": "subscriptionId",
+ "resourceGroup": DEFAULT_RESOURCE_GROUP,
+ "factory": DEFAULT_FACTORY,
+ }
+ ),
+ )
+
+ db.merge_conn(connection)
+
+
+@fixture
+def hook():
+ with patch("airflow.providers.microsoft.azure.hooks.base_azure.ServicePrincipalCredentials"):
+ with patch("airflow.providers.microsoft.azure.hooks.azure_data_factory.DataFactoryManagementClient"):
+ return AzureDataFactoryHook(conn_id="azure_data_factory_test")
+
+
+def test_hook_initialization(hook: AzureDataFactoryHook):
+ assert hook._resource_group_name == DEFAULT_RESOURCE_GROUP
+ assert hook._factory_name == DEFAULT_FACTORY
+
+
+def test_targeted_factory(hook: AzureDataFactoryHook):
+ expected = (RESOURCE_GROUP, FACTORY)
+
+ result = hook._get_targeted_factory(*expected)
+
+ assert result == expected
+
+
+def test_targeted_factory_defaults(hook: AzureDataFactoryHook):
+ expected = (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY)
+
+ result = hook._get_targeted_factory(None, None)
+
+ assert result == expected
Review comment:
For tests with different input and/or output parameters and basically have the same testing logic [parameterized](https://pypi.org/project/parameterized/) tests is a great tool :)
See [this](https://github.com/apache/airflow/blob/9549274d110f689a0bd709db829a4d69e274eed9/tests/providers/google/cloud/operators/test_cloud_build.py#L48-L67) for example.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-783216440
[The Workflow run](https://github.com/apache/airflow/actions/runs/588598340) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ShawnMcGough commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
ShawnMcGough commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-786124835
@kaxil - does this need to be re-queued or just waiting for the reviewers?
Thanks!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514955151
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I'm glad you learned something from me 😋
Unfortunately, when you try to access something that doesn't exist it fails with a "403 - Unauthorized":
```
The client 'xxxx' with object id 'xxxx' does not have authorization to perform action 'Microsoft.DataFactory/factories/read' over scope '/subscriptions/xxxx/resourceGroups/xxxx/providers/Microsoft.DataFactory/factories/xxxx' or the scope is invalid. If access was recently granted, please refresh your credentials.
```
This could also be the case if the user doesn't set his credentials correctly. 😕
[Edit]
Thinking a bit more about it, if there is something wrong with the user authorizations the second call would anyway fail with the same error so this is not misleading the user.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-721707421
[The Workflow run](https://github.com/apache/airflow/actions/runs/345529397) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514453558
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,658 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514955151
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I'm glad you learned something from me 😋
Unfortunately, when you try to access something that doesn't exist it fails with a "403 - Unauthorized":
```
The client 'xxxx' with object id 'xxxx' does not have authorization to perform action 'Microsoft.DataFactory/factories/read' over scope '/subscriptions/xxxx/resourceGroups/xxxx/providers/Microsoft.DataFactory/factories/xxxx' or the scope is invalid. If access was recently granted, please refresh your credentials.
```
This could also be the case if the user doesn't set his credentials correctly. 😕
[Edit]
Thinking a bit more about it, if there is something wrong with the user authorizations the second call would anyway fail with the same error.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] RosterIn commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
RosterIn commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-770097775
There are still some static failures with pylint
```
************* Module tests.providers.microsoft.azure.hooks.test_azure_data_factory
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:42:17: W0613: Unused argument 'module' (unused-argument)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:95:4: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:91:13: W0613: Unused argument 'self' (unused-argument)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:113:21: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:123:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:133:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:143:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:153:28: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:163:31: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:173:31: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:183:31: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:193:21: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:203:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:213:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:223:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:233:22: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:243:25: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:253:25: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:263:25: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:273:22: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:283:26: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:293:29: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:303:21: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:313:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:323:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:333:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:343:23: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:353:22: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:363:23: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:373:24: W0621: Redefining name 'hook' from outer scope (line 62) (redefined-outer-name)
************* Module airflow.providers.microsoft.azure.hooks.azure_data_factory
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:72:0: R0904: Too many public methods (28/27) (too-many-public-methods)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-712228951
Thank you very much @feluelle for your review. I've applied all your comments. Note that `parameterized` seems to be incompatible with the latest versions of `pytest`. What do you think of using `@pytest.mark.parametrize` instead?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-712290442
Tests are ignored and pytest emits a `PytestCollectionWarning` stating that "yield tests are no more collected".
How do you see the tests being refactored? I had in mind one test per method and use parameterization for the implicit and explicit factory cases. It Is also possible to go far and have one or two hyper parameterized functions but I think it will make the tests less readable. What do you think?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-715486269
[The Workflow run](https://github.com/apache/airflow/actions/runs/324638421) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-721446950
Can you please rebase your PR on latest Master since we have applied [Black](https://github.com/apache/airflow/commit/4e8f9cc8d02b29c325b8a5a76b4837671bdf5f68) and [PyUpgrade](https://github.com/apache/airflow/commit/8c42cf1b00c90f0d7f11b8a3a455381de8e003c5) on Master.
It will help if your squash your commits into single commit first so that there are less conflicts.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-785308185
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest master at your convenience, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r510758299
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I did it that way so that we can potentially add existence checks to the `create` methods and throw an exception when the entity already exists. But that could also be left to the user code.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-785272815
Let's wait for the CI before we merge 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil merged pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil merged pull request #11015:
URL: https://github.com/apache/airflow/pull/11015
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r501609552
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection during `__init__` because init will be called every time the dag file will be parsed - more than you actually need a connection to it. You only want it to be called when an actual task runs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r501609552
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection in during `__init__` because init will be called every time the dag file will be parsed - more than it actually needs. You only want it to be called when an actual task runs.
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ """
+ Return a Data Factory client.
+ """
+
+ conn = self.get_connection(self.conn_id)
+
+ self._resource_group_name = conn.extra_dejson.get("resourceGroup")
+ self._factory_name = conn.extra_dejson.get("factory")
+
+ return super().get_conn()
+
+ def _get_targeted_factory(
+ self, resource_group_name: Optional[str], factory_name: Optional[str]
+ ) -> Tuple[str, str]:
Review comment:
You could also implement this as an annotation / decorator.
Take a look at s3 for example:
- The creation: https://github.com/apache/airflow/blob/master/airflow/providers/amazon/aws/hooks/s3.py#L45-L65
- The call: https://github.com/apache/airflow/blob/master/airflow/providers/amazon/aws/hooks/s3.py#L140
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ """
+ Return a Data Factory client.
+ """
+
+ conn = self.get_connection(self.conn_id)
+
+ self._resource_group_name = conn.extra_dejson.get("resourceGroup")
+ self._factory_name = conn.extra_dejson.get("factory")
Review comment:
WDYT of putting this into `_get_targeted_factory` ?
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection during `__init__` because init will be called every time the dag file will be parsed - more than it actually needs. You only want it to be called when an actual task runs.
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection during `__init__` because init will be called every time the dag file will be parsed - more than you actually need a connection to it. You only want it to be called when an actual task runs.
##########
File path: tests/providers/microsoft/azure/hooks/test_azure_data_factory.py
##########
@@ -0,0 +1,440 @@
+# 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 re import A
+from unittest.mock import patch
+
+import pytest
+from pytest import fixture
+
+from airflow.exceptions import AirflowException
+from airflow.models.connection import Connection
+from airflow.providers.microsoft.azure.hooks.azure_data_factory import AzureDataFactoryHook
+from airflow.utils import db
+
+
+DEFAULT_RESOURCE_GROUP = "defaultResourceGroup"
+RESOURCE_GROUP = "testResourceGroup"
+
+DEFAULT_FACTORY = "defaultFactory"
+FACTORY = "testFactory"
+
+MODEL = object()
+
+
+def setup_module(module):
+ connection = Connection(
+ conn_id="azure_data_factory_test",
+ conn_type="azure_data_factory",
+ login="clientId",
+ password="clientSecret",
+ extra=json.dumps(
+ {
+ "tenantId": "tenantId",
+ "subscriptionId": "subscriptionId",
+ "resourceGroup": DEFAULT_RESOURCE_GROUP,
+ "factory": DEFAULT_FACTORY,
+ }
+ ),
+ )
+
+ db.merge_conn(connection)
+
+
+@fixture
+def hook():
+ with patch("airflow.providers.microsoft.azure.hooks.base_azure.ServicePrincipalCredentials"):
+ with patch("airflow.providers.microsoft.azure.hooks.azure_data_factory.DataFactoryManagementClient"):
+ return AzureDataFactoryHook(conn_id="azure_data_factory_test")
+
+
+def test_hook_initialization(hook: AzureDataFactoryHook):
+ assert hook._resource_group_name == DEFAULT_RESOURCE_GROUP
+ assert hook._factory_name == DEFAULT_FACTORY
+
+
+def test_targeted_factory(hook: AzureDataFactoryHook):
+ expected = (RESOURCE_GROUP, FACTORY)
+
+ result = hook._get_targeted_factory(*expected)
+
+ assert result == expected
+
+
+def test_targeted_factory_defaults(hook: AzureDataFactoryHook):
+ expected = (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY)
+
+ result = hook._get_targeted_factory(None, None)
+
+ assert result == expected
Review comment:
For tests with different input and/or output parameters and basically have the same testing logic [parameterized](https://pypi.org/project/parameterized/) tests is a great tool :)
See [this](https://github.com/apache/airflow/blob/9549274d110f689a0bd709db829a4d69e274eed9/tests/providers/google/cloud/operators/test_cloud_build.py#L48-L67) for example.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] RosterIn commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
RosterIn commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r566831231
##########
File path: airflow/providers/microsoft/azure/provider.yaml
##########
@@ -59,6 +59,10 @@ integrations:
external-doc-url: https://cloud.google.com/storage/
logo: /integration-logos/azure/Microsoft-Azure-Fileshare.png
tags: [azure]
+ - integration-name: Microsoft Azure Data Factory
+ external-doc-url: https://azure.microsoft.com/en-us/services/data-factory/
+ logo: /integration-logos/azure/Azure Data Factory.svg
+ tags: [azure]
Review comment:
```suggestion
tags: [azure]
```
Static check is failing because of whitespace
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] don4of4 commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
don4of4 commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-782928423
@flvndh this looks very close -- will you be getting a chance to look at these failing tests?
```
=================================== FAILURES ===================================
____________________ test_update_factory[explicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8215b93c8>
user_args = (<object object at 0x7ff853117360>, 'testResourceGroup', 'testFactory')
sdk_args = ('testResourceGroup', 'testFactory', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, MODEL)),
implicit_factory=((MODEL,), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, MODEL)),
)
def test_update_factory(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_factory(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:137:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8215b93c8>
factory = <object object at 0x7ff853117360>
resource_group_name = 'testResourceGroup', factory_name = 'testFactory'
config = {}
@provide_targeted_factory
def update_factory(
self,
factory: Factory,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> Factory:
"""
Update the factory.
:param factory: The factory resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the factory does not exist.
:return: The factory.
"""
if not self._factory_exists(resource_group_name, factory):
> raise AirflowException(f"Factory {factory!r} does not exist.")
E airflow.exceptions.AirflowException: Factory <object object at 0x7ff853117360> does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:130: AirflowException
____________________ test_update_factory[implicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8215874e0>
user_args = (<object object at 0x7ff853117360>,)
sdk_args = ('defaultResourceGroup', 'defaultFactory', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, MODEL)),
implicit_factory=((MODEL,), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, MODEL)),
)
def test_update_factory(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_factory(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:137:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8215874e0>
factory = <object object at 0x7ff853117360>
resource_group_name = 'defaultResourceGroup', factory_name = 'defaultFactory'
config = {}
@provide_targeted_factory
def update_factory(
self,
factory: Factory,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> Factory:
"""
Update the factory.
:param factory: The factory resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the factory does not exist.
:return: The factory.
"""
if not self._factory_exists(resource_group_name, factory):
> raise AirflowException(f"Factory {factory!r} does not exist.")
E airflow.exceptions.AirflowException: Factory <object object at 0x7ff853117360> does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:130: AirflowException
_________________ test_update_linked_service[explicit factory] _________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8202feef0>
user_args = ('testName', <object object at 0x7ff853117360>, 'testResourceGroup', 'testFactory')
sdk_args = ('testResourceGroup', 'testFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_linked_service(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_linked_service(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:177:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8202feef0>
linked_service_name = 'testName'
linked_service = <object object at 0x7ff853117360>
resource_group_name = 'testResourceGroup', factory_name = 'testFactory'
config = {}
@provide_targeted_factory
def update_linked_service(
self,
linked_service_name: str,
linked_service: LinkedService,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> LinkedServiceResource:
"""
Update the linked service.
:param linked_service_name: The linked service name.
:param linked_service: The linked service resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the linked service does not exist.
:return: The linked service.
"""
if not self._linked_service_exists(resource_group_name, factory_name, linked_service_name):
> raise AirflowException(f"Linked service {linked_service_name!r} does not exist.")
E airflow.exceptions.AirflowException: Linked service 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:227: AirflowException
_________________ test_update_linked_service[implicit factory] _________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8202fe710>
user_args = ('testName', <object object at 0x7ff853117360>)
sdk_args = ('defaultResourceGroup', 'defaultFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_linked_service(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_linked_service(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:177:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8202fe710>
linked_service_name = 'testName'
linked_service = <object object at 0x7ff853117360>
resource_group_name = 'defaultResourceGroup', factory_name = 'defaultFactory'
config = {}
@provide_targeted_factory
def update_linked_service(
self,
linked_service_name: str,
linked_service: LinkedService,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> LinkedServiceResource:
"""
Update the linked service.
:param linked_service_name: The linked service name.
:param linked_service: The linked service resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the linked service does not exist.
:return: The linked service.
"""
if not self._linked_service_exists(resource_group_name, factory_name, linked_service_name):
> raise AirflowException(f"Linked service {linked_service_name!r} does not exist.")
E airflow.exceptions.AirflowException: Linked service 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:227: AirflowException
____________________ test_update_dataset[explicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff829e30198>
user_args = ('testName', <object object at 0x7ff853117360>, 'testResourceGroup', 'testFactory')
sdk_args = ('testResourceGroup', 'testFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_dataset(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_dataset(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:217:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff829e30198>
dataset_name = 'testName', dataset = <object object at 0x7ff853117360>
resource_group_name = 'testResourceGroup', factory_name = 'testFactory'
config = {}
@provide_targeted_factory
def update_dataset(
self,
dataset_name: str,
dataset: Dataset,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> DatasetResource:
"""
Update the dataset.
:param dataset_name: The dataset name.
:param dataset: The dataset resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the dataset does not exist.
:return: The dataset.
"""
if not self._dataset_exists(resource_group_name, factory_name, dataset_name):
> raise AirflowException(f"Dataset {dataset_name!r} does not exist.")
E airflow.exceptions.AirflowException: Dataset 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:329: AirflowException
____________________ test_update_dataset[implicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff82a063400>
user_args = ('testName', <object object at 0x7ff853117360>)
sdk_args = ('defaultResourceGroup', 'defaultFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_dataset(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_dataset(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:217:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff82a063400>
dataset_name = 'testName', dataset = <object object at 0x7ff853117360>
resource_group_name = 'defaultResourceGroup', factory_name = 'defaultFactory'
config = {}
@provide_targeted_factory
def update_dataset(
self,
dataset_name: str,
dataset: Dataset,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> DatasetResource:
"""
Update the dataset.
:param dataset_name: The dataset name.
:param dataset: The dataset resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the dataset does not exist.
:return: The dataset.
"""
if not self._dataset_exists(resource_group_name, factory_name, dataset_name):
> raise AirflowException(f"Dataset {dataset_name!r} does not exist.")
E airflow.exceptions.AirflowException: Dataset 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:329: AirflowException
____________________ test_update_pipeline[explicit factory] ____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8231ae198>
user_args = ('testName', <object object at 0x7ff853117360>, 'testResourceGroup', 'testFactory')
sdk_args = ('testResourceGroup', 'testFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_pipeline(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_pipeline(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:257:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff8231ae198>
pipeline_name = 'testName', pipeline = <object object at 0x7ff853117360>
resource_group_name = 'testResourceGroup', factory_name = 'testFactory'
config = {}
@provide_targeted_factory
def update_pipeline(
self,
pipeline_name: str,
pipeline: PipelineResource,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> PipelineResource:
"""
Update the pipeline.
:param pipeline_name: The pipeline name.
:param pipeline: The pipeline resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the pipeline does not exist.
:return: The pipeline.
"""
if not self._pipeline_exists(resource_group_name, factory_name, pipeline_name):
> raise AirflowException(f"Pipeline {pipeline_name!r} does not exist.")
E airflow.exceptions.AirflowException: Pipeline 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:429: AirflowException
____________________ test_update_pipeline[implicit factory] ____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81bf956d8>
user_args = ('testName', <object object at 0x7ff853117360>)
sdk_args = ('defaultResourceGroup', 'defaultFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_pipeline(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_pipeline(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:257:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81bf956d8>
pipeline_name = 'testName', pipeline = <object object at 0x7ff853117360>
resource_group_name = 'defaultResourceGroup', factory_name = 'defaultFactory'
config = {}
@provide_targeted_factory
def update_pipeline(
self,
pipeline_name: str,
pipeline: PipelineResource,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> PipelineResource:
"""
Update the pipeline.
:param pipeline_name: The pipeline name.
:param pipeline: The pipeline resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the pipeline does not exist.
:return: The pipeline.
"""
if not self._pipeline_exists(resource_group_name, factory_name, pipeline_name):
> raise AirflowException(f"Pipeline {pipeline_name!r} does not exist.")
E airflow.exceptions.AirflowException: Pipeline 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:429: AirflowException
____________________ test_update_trigger[explicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81adab5f8>
user_args = ('testName', <object object at 0x7ff853117360>, 'testResourceGroup', 'testFactory')
sdk_args = ('testResourceGroup', 'testFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_trigger(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_trigger(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:327:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81adab5f8>
trigger_name = 'testName', trigger = <object object at 0x7ff853117360>
resource_group_name = 'testResourceGroup', factory_name = 'testFactory'
config = {}
@provide_targeted_factory
def update_trigger(
self,
trigger_name: str,
trigger: Trigger,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> TriggerResource:
"""
Update the trigger.
:param trigger_name: The trigger name.
:param trigger: The trigger resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the trigger does not exist.
:return: The trigger.
"""
if not self._trigger_exists(resource_group_name, factory_name, trigger_name):
> raise AirflowException(f"Trigger {trigger_name!r} does not exist.")
E airflow.exceptions.AirflowException: Trigger 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:587: AirflowException
____________________ test_update_trigger[implicit factory] _____________________
hook = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81adabc18>
user_args = ('testName', <object object at 0x7ff853117360>)
sdk_args = ('defaultResourceGroup', 'defaultFactory', 'testName', <object object at 0x7ff853117360>)
@parametrize(
explicit_factory=((NAME, MODEL, RESOURCE_GROUP, FACTORY), (RESOURCE_GROUP, FACTORY, NAME, MODEL)),
implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
)
def test_update_trigger(hook: AzureDataFactoryHook, user_args, sdk_args):
> hook.update_trigger(*user_args)
tests/providers/microsoft/azure/hooks/test_azure_data_factory.py:327:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:67: in wrapper
return func(*bound_args.args, **bound_args.kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airflow.providers.microsoft.azure.hooks.azure_data_factory.AzureDataFactoryHook object at 0x7ff81adabc18>
trigger_name = 'testName', trigger = <object object at 0x7ff853117360>
resource_group_name = 'defaultResourceGroup', factory_name = 'defaultFactory'
config = {}
@provide_targeted_factory
def update_trigger(
self,
trigger_name: str,
trigger: Trigger,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> TriggerResource:
"""
Update the trigger.
:param trigger_name: The trigger name.
:param trigger: The trigger resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:raise AirflowException: If the trigger does not exist.
:return: The trigger.
"""
if not self._trigger_exists(resource_group_name, factory_name, trigger_name):
> raise AirflowException(f"Trigger {trigger_name!r} does not exist.")
E airflow.exceptions.AirflowException: Trigger 'testName' does not exist.
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:587: AirflowException
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] don4of4 commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
don4of4 commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-777576086
What is the status of this PR? How can others help get this across the line?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-712279324
Oh really? What error do you get? `@pytest.mark.parametrize` seems also to be a good option to use :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-783292282
@don4of4 Here you go 😀
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil edited a comment on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil edited a comment on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-777742610
Tests are failing:
```
0.58s call tests/providers/amazon/aws/hooks/test_s3.py::TestAwsS3Hook::test_get_wildcard_key
=========================== short test summary info ============================
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_factory[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_factory[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_linked_service[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_linked_service[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_dataset[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_dataset[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_pipeline[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_pipeline[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_trigger[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_trigger[implicit factory]
= 10 failed, 3977 passed, 191 skipped, 3 xfailed, 2 warnings in 320.25s (0:05:20) =
```
https://github.com/apache/airflow/pull/11015/checks?check_run_id=1881944784#step:6:3230
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-715136031
> I had in mind one test per method and use parameterization for the implicit and explicit factory cases.
I like this. 👍
> It Is also possible to go far and have one or two hyper parameterized functions but I think it will make the tests less readable.
Readability is far more important than code duplication in my opinion. So I am in favor for one test per method.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514452717
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I had something like this in mind for the checks:
```python
@provide_targeted_factory
def create_factory(
self,
factory: Factory,
resource_group_name: Optional[str] = None,
factory_name: Optional[str] = None,
**config: Any,
) -> Factory:
"""
Create the factory.
:param factory: The factory resource definition.
:param resource_group_name: The resource group name.
:param factory_name: The factory name.
:param config: Extra parameters for the ADF client.
:return: The factory.
"""
factories = self.get_conn().factories
try:
factories.get(resource_group_name, factory_name)
except CloudError:
return factories.create_or_update(resource_group_name, factory_name, factory, **config)
else:
raise AirflowException(f"Factory {factory_name!r} already exists")
```
but this `CloudError` is a little too generic to my taste. What would you do ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-770997205
I would say, let's ignore the error i.e. sth. like: `# pylint: disable=too-many-public-methods` for this specific module.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-770722905
@RosterIn Most of the warnings are caused by the pytest [setup](https://docs.pytest.org/en/stable/xunit_setup.html#module-level-setup-teardown) and [fixture](https://docs.pytest.org/en/stable/fixture.html#requesting-fixtures).
For the `Too many public methods (28/27)` I could move related methods into their own object and the API could look like this:
```python
>>> hook.factory.get(...)
>>> hook.factory.create(...)
>>> hook.trigger.update(...)
>>> hook.dataset.delete(...)
```
but this is quite close to what the Microsoft SDK does. @feluelle what's your view on that ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r510713489
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
There are many `create_or_update`s. Same goes for them.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-783218636
[The Workflow run](https://github.com/apache/airflow/actions/runs/588604172) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-786751214
> @kaxil looks like a doc build failed and two MySQL/Py builds. 🤔
Transient failures :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] don4of4 commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
don4of4 commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-785195417
@kaxil Can we get this reviewed?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-769769689
[The Workflow run](https://github.com/apache/airflow/actions/runs/520775456) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r510713489
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
There are many other `create_or_update`s. Same goes for them.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514955151
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I'm glad you learned something from me 😋
Unfortunately, when you try to access something that doesn't exist it fails with a "403 - Unauthorized":
```
The client 'xxxx' with object id 'xxxx' does not have authorization to perform action 'Microsoft.DataFactory/factories/read' over scope '/subscriptions/xxxx/resourceGroups/xxxx/providers/Microsoft.DataFactory/factories/xxxx' or the scope is invalid. If access was recently granted, please refresh your credentials.
```
This could also be the case if the user doesn't set his credentials correctly. 😕
[Edit]
Thinking a bit more about it, if there is something wrong with the user authorizations the second call would anyway fail with the same error so this is not misleading the user. Thus, problem solved for the `create` method but the same applies if we want to check that the factory exists when we call `update`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mivinson commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
mivinson commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-769410252
@flvndh Can you please look at the merge conflicts? This is a very useful package that you have authored.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-721366724
[The Workflow run](https://github.com/apache/airflow/actions/runs/344335912) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] stale[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
stale[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-751239955
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-777742610
Tests are failing:
```
0.58s call tests/providers/amazon/aws/hooks/test_s3.py::TestAwsS3Hook::test_get_wildcard_key
=========================== short test summary info ============================
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_factory[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_factory[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_linked_service[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_linked_service[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_dataset[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_dataset[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_pipeline[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_pipeline[implicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_trigger[explicit factory]
FAILED tests/providers/microsoft/azure/hooks/test_azure_data_factory.py::test_update_trigger[implicit factory]
= 10 failed, 3977 passed, 191 skipped, 3 xfailed, 2 warnings in 320.25s (0:05:20) =
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] don4of4 edited a comment on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
don4of4 edited a comment on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-782928423
@flvndh this looks very close -- will you be getting a chance to look at these failing tests?
```
=================================== FAILURES ===================================
____________________ test_update_factory[explicit factory] _____________________
-trimmed-
airflow/providers/microsoft/azure/hooks/azure_data_factory.py:587: AirflowException
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r514955151
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I'm glad you learned something from me 😋
Unfortunately, when you try to access something that doesn't exist it fails with a "403 - Unauthorized":
```
The client 'xxxx' with object id 'xxxx' does not have authorization to perform action 'Microsoft.DataFactory/factories/read' over scope '/subscriptions/xxxx/resourceGroups/xxxx/providers/Microsoft.DataFactory/factories/xxxx' or the scope is invalid. If access was recently granted, please refresh your credentials.
```
This could also be the case if the user doesn't set his credentials correctly. 😕
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ShawnMcGough commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
ShawnMcGough commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-786747118
@kaxil looks like a doc build failed and two MySQL/Py builds. 🤔
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r513258628
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,658 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
Review comment:
Could you also add test for this method, please?
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,659 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
+ """
+ Provide the targeted factory to the decorated function in case it isn't specified.
+
+ If ``resource_group_name`` or ``factory_name`` is not provided it defaults to the value specified in
+ the connection extras.
+ """
+
+ signature = inspect.signature(func)
+
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Callable:
+ bound_args = signature.bind(*args, **kwargs)
+
+ def bind_argument(arg, default_key):
+ if arg not in bound_args.arguments:
+ self = args[0]
+ conn = self.get_connection(self.conn_id)
+ default_value = conn.extra_dejson.get(default_key)
+
+ if not default_value:
+ raise AirflowException("Could not determine the targeted data factory.")
+
+ bound_args.arguments[arg] = conn.extra_dejson[default_key]
+
+ bind_argument("resource_group_name", "resourceGroup")
+ bind_argument("factory_name", "factory")
+
+ return func(*bound_args.args, **bound_args.kwargs)
+
+ return wrapper
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+ self._conn = None
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ if not self._conn:
+ self._conn = super().get_conn()
+
+ return self._conn
+
+ @provide_targeted_factory
+ def get_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> Factory:
+ """
+ Get the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.get(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def update_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Update the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def create_factory(
+ self,
+ factory: Factory,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> Factory:
+ """
+ Create the factory.
+
+ :param factory: The factory resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The factory.
+ """
+
+ return self.get_conn().factories.create_or_update(
+ resource_group_name, factory_name, factory, **config
+ )
+
+ @provide_targeted_factory
+ def delete_factory(
+ self, resource_group_name: Optional[str] = None, factory_name: Optional[str] = None, **config: Any
+ ) -> None:
+ """
+ Delete the factory.
+
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().factories.delete(resource_group_name, factory_name, **config)
+
+ @provide_targeted_factory
+ def get_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Get the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.get(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def update_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Update the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def create_linked_service(
+ self,
+ linked_service_name: str,
+ linked_service: LinkedService,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> LinkedServiceResource:
+ """
+ Create the linked service.
+
+ :param linked_service_name: The linked service name.
+ :param linked_service: The linked service resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The linked service.
+ """
+
+ return self.get_conn().linked_services.create_or_update(
+ resource_group_name, factory_name, linked_service_name, linked_service, **config
+ )
+
+ @provide_targeted_factory
+ def delete_linked_service(
+ self,
+ linked_service_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> None:
+ """
+ Delete the linked service:
+
+ :param linked_service_name: The linked service name.
+ :param resource_group_name: The linked service name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ """
+
+ self.get_conn().linked_services.delete(
+ resource_group_name, factory_name, linked_service_name, **config
+ )
+
+ @provide_targeted_factory
+ def get_dataset(
+ self,
+ dataset_name: str,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Get the dataset.
+
+ :param dataset_name: The dataset name.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.get(resource_group_name, factory_name, dataset_name, **config)
+
+ @provide_targeted_factory
+ def update_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Update the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
+
+ @provide_targeted_factory
+ def create_dataset(
+ self,
+ dataset_name: str,
+ dataset: Dataset,
+ resource_group_name: Optional[str] = None,
+ factory_name: Optional[str] = None,
+ **config: Any,
+ ) -> DatasetResource:
+ """
+ Create the dataset.
+
+ :param dataset_name: The dataset name.
+ :param dataset: The dataset resource definition.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The factory name.
+ :param config: Extra parameters for the ADF client.
+ :return: The dataset.
+ """
+
+ return self.get_conn().datasets.create_or_update(
+ resource_group_name, factory_name, dataset_name, dataset, **config
+ )
Review comment:
I see. Can you add this into the method instead?
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,658 @@
+# 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 inspect
+from functools import wraps
+from typing import Any, Callable, Optional
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+def provide_targeted_factory(func: Callable) -> Callable:
Review comment:
The tests do not cover this (functionality of using the connection to fill in the params)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-785406913
[The Workflow run](https://github.com/apache/airflow/actions/runs/597485650) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] flvndh commented on pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
flvndh commented on pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#issuecomment-769866477
> @flvndh Can you please resolve the conflicts and ping us again, thanks
@kaxil conflicts were resolved and the branch was rebased on the latest master 😉
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #11015: Add Azure Data Factory hook
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #11015:
URL: https://github.com/apache/airflow/pull/11015#discussion_r501609552
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
Review comment:
```suggestion
```
Please don't establish a connection in during `__init__` because init will be called every time the dag file will be parsed - more than it actually needs. You only want it to be called when an actual task runs.
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ """
+ Return a Data Factory client.
+ """
+
+ conn = self.get_connection(self.conn_id)
+
+ self._resource_group_name = conn.extra_dejson.get("resourceGroup")
+ self._factory_name = conn.extra_dejson.get("factory")
+
+ return super().get_conn()
+
+ def _get_targeted_factory(
+ self, resource_group_name: Optional[str], factory_name: Optional[str]
+ ) -> Tuple[str, str]:
Review comment:
You could also implement this as an annotation / decorator.
Take a look at s3 for example:
- The creation: https://github.com/apache/airflow/blob/master/airflow/providers/amazon/aws/hooks/s3.py#L45-L65
- The call: https://github.com/apache/airflow/blob/master/airflow/providers/amazon/aws/hooks/s3.py#L140
##########
File path: airflow/providers/microsoft/azure/hooks/azure_data_factory.py
##########
@@ -0,0 +1,674 @@
+# 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 typing import Any, Optional, Tuple
+
+from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models import (
+ CreateRunResponse,
+ Dataset,
+ DatasetResource,
+ Factory,
+ LinkedService,
+ LinkedServiceResource,
+ PipelineResource,
+ PipelineRun,
+ Trigger,
+ TriggerResource,
+)
+from msrestazure.azure_operation import AzureOperationPoller
+
+from airflow.exceptions import AirflowException
+from airflow.providers.microsoft.azure.hooks.base_azure import AzureBaseHook
+
+
+class AzureDataFactoryHook(AzureBaseHook):
+ """
+ A hook to interact with Azure Data Factory.
+
+ :param conn_id: The Azure Data Factory connection id.
+ """
+
+ def __init__(self, conn_id: str = "azure_data_factory_default"):
+ super().__init__(sdk_client=DataFactoryManagementClient, conn_id=conn_id)
+
+ self._conn = self.get_conn()
+
+ def get_conn(self) -> DataFactoryManagementClient:
+ """
+ Return a Data Factory client.
+ """
+
+ conn = self.get_connection(self.conn_id)
+
+ self._resource_group_name = conn.extra_dejson.get("resourceGroup")
+ self._factory_name = conn.extra_dejson.get("factory")
Review comment:
WDYT of putting this into `_get_targeted_factory` ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org