This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6ba529e0ebf Add dedicated Glue crawler lifecycle operators (#69930)
6ba529e0ebf is described below
commit 6ba529e0ebf0bd829bed78201a89b3a59ec809df
Author: Morgan <[email protected]>
AuthorDate: Thu Jul 16 21:53:23 2026 -0300
Add dedicated Glue crawler lifecycle operators (#69930)
The existing Glue crawler task combines resource provisioning,
configuration changes, and execution. This makes routine crawler runs capable
of mutating infrastructure and obscures the permissions each task actually
needs.
Split the AWS Glue crawler lifecycle into explicit create, update, run, and
delete operators while keeping the existing GlueCrawlerOperator backward
compatible during its deprecation period. Users need at least one full major
provider release cycle to migrate from this widely used operator.
---
.../tests/unit/always/test_project_structure.py | 2 +
providers/amazon/docs/operators/glue.rst | 81 ++++++-
.../providers/amazon/aws/operators/glue_crawler.py | 255 +++++++++++++++++----
.../amazon/tests/system/amazon/aws/example_glue.py | 75 +++++-
.../unit/amazon/aws/operators/test_glue_crawler.py | 198 +++++++++++++++-
5 files changed, 539 insertions(+), 72 deletions(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index 750966a3202..9286bdb5e41 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -535,6 +535,7 @@ class
TestAmazonProviderProjectStructure(ExampleCoverageTest):
BASE_CLASSES = {
"airflow.providers.amazon.aws.operators.base_aws.AwsBaseOperator",
+
"airflow.providers.amazon.aws.operators.glue_crawler._GlueCrawlerBaseOperator",
"airflow.providers.amazon.aws.operators.rds.RdsBaseOperator",
"airflow.providers.amazon.aws.operators.sagemaker.SageMakerBaseOperator",
"airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor",
@@ -564,6 +565,7 @@ class
TestAmazonProviderProjectStructure(ExampleCoverageTest):
}
DEPRECATED_CLASSES = {
+
"airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerOperator",
"airflow.providers.amazon.aws.operators.lambda_function.AwsLambdaInvokeFunctionOperator",
}
diff --git a/providers/amazon/docs/operators/glue.rst
b/providers/amazon/docs/operators/glue.rst
index c3e156c4dc8..e4a9466ba22 100644
--- a/providers/amazon/docs/operators/glue.rst
+++ b/providers/amazon/docs/operators/glue.rst
@@ -37,26 +37,97 @@ Generic Parameters
Operators
---------
-.. _howto/operator:GlueCrawlerOperator:
+.. _howto/operator:GlueCrawlerCreateOperator:
Create an AWS Glue crawler
==========================
AWS Glue Crawlers allow you to easily extract data from various data sources.
-To create a new AWS Glue Crawler or run an existing one you can
-use
:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerOperator`.
+To create a crawler, use
+:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerCreateOperator`.
.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_glue.py
:language: python
:dedent: 4
- :start-after: [START howto_operator_glue_crawler]
- :end-before: [END howto_operator_glue_crawler]
+ :start-after: [START howto_operator_glue_crawler_create]
+ :end-before: [END howto_operator_glue_crawler_create]
.. note::
The AWS IAM role included in the ``config`` needs access to the source data
location
(e.g. s3:PutObject access if data is stored in Amazon S3) as well as the
``AWSGlueServiceRole``
policy. See the References section below for a link to more details.
+.. _howto/operator:GlueCrawlerUpdateOperator:
+
+Update an AWS Glue crawler
+==========================
+
+To update the configuration of an existing crawler, use
+:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerUpdateOperator`.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_glue.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_glue_crawler_update]
+ :end-before: [END howto_operator_glue_crawler_update]
+
+.. _howto/operator:GlueCrawlerRunOperator:
+
+Run an AWS Glue crawler
+=======================
+
+To run an existing crawler and wait for it to complete, use
+:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerRunOperator`.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_glue.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_glue_crawler_run]
+ :end-before: [END howto_operator_glue_crawler_run]
+
+The operator waits for completion by default. Set ``deferrable=True`` to
perform the wait without
+occupying a worker slot.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_glue.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_glue_crawler_run_deferrable]
+ :end-before: [END howto_operator_glue_crawler_run_deferrable]
+
+.. _howto/operator:GlueCrawlerDeleteOperator:
+
+Delete an AWS Glue crawler
+==========================
+
+To delete an existing crawler, use
+:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerDeleteOperator`.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_glue.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_glue_crawler_delete]
+ :end-before: [END howto_operator_glue_crawler_delete]
+
+.. _howto/operator:GlueCrawlerOperator:
+
+Legacy AWS Glue crawler operator
+================================
+
+.. warning::
+
:class:`~airflow.providers.amazon.aws.operators.glue_crawler.GlueCrawlerOperator`
is deprecated.
+ Existing Dags can continue using it during the deprecation period, but new
Dags should use the
+ operation-specific operators above.
+
+The legacy operator creates or updates a crawler and then runs it. Existing
Dags can continue using
+the same configuration while migrating each operation to the dedicated
operators:
+
+.. code-block:: python
+
+ crawl_s3 = GlueCrawlerOperator(
+ task_id="crawl_s3",
+ config=glue_crawler_config,
+ )
+
.. _howto/operator:GlueJobOperator:
Submit an AWS Glue job
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_crawler.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_crawler.py
index 7c8e0d54d5b..29ad429276f 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_crawler.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_crawler.py
@@ -17,11 +17,13 @@
# under the License.
from __future__ import annotations
+import warnings
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any
from botocore.exceptions import ClientError
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.amazon.aws.hooks.glue_crawler import GlueCrawlerHook
from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
from airflow.providers.amazon.aws.triggers.glue_crawler import
GlueCrawlerCompleteTrigger
@@ -33,9 +35,80 @@ if TYPE_CHECKING:
from airflow.sdk import Context
-class GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
+class _GlueCrawlerBaseOperator(AwsBaseOperator[GlueCrawlerHook]):
+ aws_hook_class = GlueCrawlerHook
+ ui_color = "#ededed"
+
+
+class GlueCrawlerCreateOperator(_GlueCrawlerBaseOperator):
+ """
+ Create an AWS Glue crawler.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:GlueCrawlerCreateOperator`
+
+ :param config: Configurations for the AWS Glue crawler.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is ``None`` or empty then the default boto3 behaviour is used.
If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default
boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore
client. See:
+
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
"""
- Creates, updates and triggers an AWS Glue Crawler.
+
+ template_fields: Sequence[str] = aws_template_fields("config")
+
+ def __init__(self, *, config: dict[str, Any], **kwargs):
+ super().__init__(**kwargs)
+ self.config = config
+
+ def execute(self, context: Context) -> str:
+ crawler_name = self.config["Name"]
+ self.hook.create_crawler(**self.config)
+ return crawler_name
+
+
+class GlueCrawlerUpdateOperator(_GlueCrawlerBaseOperator):
+ """
+ Update an existing AWS Glue crawler.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:GlueCrawlerUpdateOperator`
+
+ :param config: Configurations for the AWS Glue crawler.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is ``None`` or empty then the default boto3 behaviour is used.
If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default
boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore
client. See:
+
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ template_fields: Sequence[str] = aws_template_fields("config")
+
+ def __init__(self, *, config: dict[str, Any], **kwargs):
+ super().__init__(**kwargs)
+ self.config = config
+
+ def execute(self, context: Context) -> str:
+ crawler_name = self.config["Name"]
+ self.hook.update_crawler(**self.config)
+ return crawler_name
+
+
+class GlueCrawlerRunOperator(_GlueCrawlerBaseOperator):
+ """
+ Run an existing AWS Glue crawler.
AWS Glue Crawler is a serverless service that manages a catalog of
metadata tables that contain the inferred schema, format and data
@@ -43,20 +116,20 @@ class
GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
.. seealso::
For more information on how to use this operator, take a look at the
guide:
- :ref:`howto/operator:GlueCrawlerOperator`
+ :ref:`howto/operator:GlueCrawlerRunOperator`
- :param config: Configurations for the AWS Glue crawler
+ :param crawler_name: Name of the AWS Glue crawler.
:param poll_interval: Time (in seconds) to wait between two consecutive
calls to check crawler status
:param wait_for_completion: Whether to wait for crawl execution
completion. (default: True)
:param deferrable: If True, the operator will wait asynchronously for the
crawl to complete.
This implies waiting for completion. This mode requires aiobotocore
module to be installed.
(default: False)
:param fail_on_already_running: If True (default), the operator will raise
an exception when
- ``start_crawler()`` or ``update_crawler()`` encounters a
``CrawlerRunningException``
- (i.e., the crawler is already running). If False, the operator logs a
warning
- and waits for the existing run to complete. Setting this to False is
useful for handling
- retry-induced race conditions where boto3 retries trigger a second
``start_crawler()``
- call after a network timeout on the first (successful) call. (default:
True)
+ ``start_crawler()`` encounters a ``CrawlerRunningException`` (i.e.,
the crawler is already
+ running). If False, the operator continues with the existing run and
waits for it if
+ ``wait_for_completion`` is True. Setting this to False is useful for
handling retry-induced race
+ conditions where boto3 retries trigger a second ``start_crawler()``
call after a network timeout
+ on the first successful call. (default: True)
:param aws_conn_id: The Airflow connection used for AWS credentials.
If this is ``None`` or empty then the default boto3 behaviour is used.
If
running Airflow in a distributed manner and aws_conn_id is None or
@@ -69,16 +142,12 @@ class
GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
"""
- aws_hook_class = GlueCrawlerHook
-
- template_fields: Sequence[str] = aws_template_fields(
- "config",
- )
- ui_color = "#ededed"
+ template_fields: Sequence[str] = aws_template_fields("crawler_name")
def __init__(
self,
- config,
+ *,
+ crawler_name: str,
poll_interval: int = 5,
wait_for_completion: bool = True,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
@@ -90,42 +159,22 @@ class
GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
self.wait_for_completion = wait_for_completion
self.deferrable = deferrable
self.fail_on_already_running = fail_on_already_running
- self.config = config
+ self.crawler_name = crawler_name
def execute(self, context: Context) -> str:
"""
- Execute AWS Glue Crawler from Airflow.
+ Run an AWS Glue crawler from Airflow.
- :return: the name of the current glue crawler.
+ :return: The name of the current Glue crawler.
"""
- crawler_name = self.config["Name"]
- if self.hook.has_crawler(crawler_name):
- try:
- self.hook.update_crawler(**self.config)
- except ClientError as e:
- if (
- not self.fail_on_already_running
- and e.response["Error"]["Code"] ==
"CrawlerRunningException"
- ):
- self.log.warning(
- "Crawler '%s' is currently running. "
- "Skipping update and waiting for the existing run to
complete.",
- crawler_name,
- )
- else:
- raise
- else:
- self.hook.create_crawler(**self.config)
-
self.log.info("Triggering AWS Glue Crawler")
try:
- self.hook.start_crawler(crawler_name)
+ self.hook.start_crawler(self.crawler_name)
except ClientError as e:
if not self.fail_on_already_running and
e.response["Error"]["Code"] == "CrawlerRunningException":
self.log.warning(
- "Crawler '%s' is already running. "
- "Waiting for the existing run to complete instead of
failing.",
- crawler_name,
+ "Crawler '%s' is already running. Continuing with the
existing run instead of failing.",
+ self.crawler_name,
)
else:
raise
@@ -133,7 +182,7 @@ class GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
if self.deferrable:
self.defer(
trigger=GlueCrawlerCompleteTrigger(
- crawler_name=crawler_name,
+ crawler_name=self.crawler_name,
waiter_delay=self.poll_interval,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
@@ -144,13 +193,129 @@ class
GlueCrawlerOperator(AwsBaseOperator[GlueCrawlerHook]):
)
elif self.wait_for_completion:
self.log.info("Waiting for AWS Glue Crawler")
- self.hook.wait_for_crawler_completion(crawler_name=crawler_name,
poll_interval=self.poll_interval)
+ self.hook.wait_for_crawler_completion(
+ crawler_name=self.crawler_name,
poll_interval=self.poll_interval
+ )
- return crawler_name
+ return self.crawler_name
def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> str:
validated_event = validate_execute_complete_event(event)
if validated_event["status"] != "success":
raise AirflowException(f"Error in glue crawl: {validated_event}")
+ return self.crawler_name
+
+
+class GlueCrawlerDeleteOperator(_GlueCrawlerBaseOperator):
+ """
+ Delete an AWS Glue crawler.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:GlueCrawlerDeleteOperator`
+
+ :param crawler_name: Name of the AWS Glue crawler.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is ``None`` or empty then the default boto3 behaviour is used.
If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default
boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore
client. See:
+
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ template_fields: Sequence[str] = aws_template_fields("crawler_name")
+
+ def __init__(self, *, crawler_name: str, **kwargs):
+ super().__init__(**kwargs)
+ self.crawler_name = crawler_name
+
+ def execute(self, context: Context) -> None:
+ self.log.info("Deleting AWS Glue crawler %s", self.crawler_name)
+ self.hook.glue_client.delete_crawler(Name=self.crawler_name)
+ self.log.info("Deleted AWS Glue crawler %s", self.crawler_name)
+
+
+# TODO: Remove GlueCrawlerOperator in Amazon provider 11.0.0 or later.
+class GlueCrawlerOperator(GlueCrawlerRunOperator):
+ """
+ Create, update and run an AWS Glue crawler.
+
+ .. deprecated::
+ Use :class:`GlueCrawlerCreateOperator`,
:class:`GlueCrawlerUpdateOperator`, and
+ :class:`GlueCrawlerRunOperator` instead.
+
+ :param config: Configurations for the AWS Glue crawler.
+ :param poll_interval: Time (in seconds) to wait between two consecutive
calls to check crawler status.
+ :param wait_for_completion: Whether to wait for crawl execution
completion. (default: True)
+ :param deferrable: If True, the operator will wait asynchronously for the
crawl to complete.
+ This implies waiting for completion. (default: False)
+ :param fail_on_already_running: If True (default), raise an exception if
the crawler is already running.
+ If False, wait for the existing run to complete.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is ``None`` or empty then the default boto3 behaviour is used.
If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default
boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore
client. See:
+
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ template_fields: Sequence[str] = aws_template_fields("config")
+
+ def __init__(
+ self,
+ config: dict[str, Any],
+ poll_interval: int = 5,
+ wait_for_completion: bool = True,
+ deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
+ fail_on_already_running: bool = True,
+ **kwargs,
+ ):
+ warnings.warn(
+ "GlueCrawlerOperator is deprecated. Use GlueCrawlerCreateOperator,
"
+ "GlueCrawlerUpdateOperator, and GlueCrawlerRunOperator instead.",
+ category=AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ super().__init__(
+ crawler_name=config["Name"],
+ poll_interval=poll_interval,
+ wait_for_completion=wait_for_completion,
+ deferrable=deferrable,
+ fail_on_already_running=fail_on_already_running,
+ **kwargs,
+ )
+ self.config = config
+
+ def execute(self, context: Context) -> str:
+ self.crawler_name = self.config["Name"]
+ if self.hook.has_crawler(self.crawler_name):
+ try:
+ self.hook.update_crawler(**self.config)
+ except ClientError as e:
+ if (
+ not self.fail_on_already_running
+ and e.response["Error"]["Code"] ==
"CrawlerRunningException"
+ ):
+ self.log.warning(
+ "Crawler '%s' is currently running. "
+ "Skipping update and waiting for the existing run to
complete.",
+ self.crawler_name,
+ )
+ else:
+ raise
+ else:
+ self.hook.create_crawler(**self.config)
+ return super().execute(context)
+
+ def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> str:
+ super().execute_complete(context, event)
return self.config["Name"]
diff --git a/providers/amazon/tests/system/amazon/aws/example_glue.py
b/providers/amazon/tests/system/amazon/aws/example_glue.py
index d62bcf3b308..b6881a5683e 100644
--- a/providers/amazon/tests/system/amazon/aws/example_glue.py
+++ b/providers/amazon/tests/system/amazon/aws/example_glue.py
@@ -25,7 +25,12 @@ if TYPE_CHECKING:
from botocore.client import BaseClient
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
-from airflow.providers.amazon.aws.operators.glue_crawler import
GlueCrawlerOperator
+from airflow.providers.amazon.aws.operators.glue_crawler import (
+ GlueCrawlerCreateOperator,
+ GlueCrawlerDeleteOperator,
+ GlueCrawlerRunOperator,
+ GlueCrawlerUpdateOperator,
+)
from airflow.providers.amazon.aws.operators.s3 import (
S3CreateBucketOperator,
S3CreateObjectOperator,
@@ -75,14 +80,21 @@
datasource.toDF().write.format('csv').mode("append").save('s3://{bucket_name}/ou
@task(trigger_rule=TriggerRule.ALL_DONE)
-def glue_cleanup(crawler_name: str, job_name: str, db_name: str) -> None:
+def delete_glue_resources(job_name: str, db_name: str) -> None:
client: BaseClient = boto3.client("glue")
- client.delete_crawler(Name=crawler_name)
client.delete_job(JobName=job_name)
client.delete_database(Name=db_name)
+@task
+def verify_crawler_description(crawler_name: str, expected_description: str)
-> None:
+ client: BaseClient = boto3.client("glue")
+
+ crawler = client.get_crawler(Name=crawler_name)["Crawler"]
+ assert crawler["Description"] == expected_description
+
+
with DAG(
dag_id=DAG_ID,
schedule="@once",
@@ -101,10 +113,16 @@ with DAG(
glue_crawler_config = {
"Name": glue_crawler_name,
+ "Description": "Glue crawler created by the Amazon provider system
test",
"Role": role_arn,
"DatabaseName": glue_db_name,
"Targets": {"S3Targets": [{"Path": f"{bucket_name}/input"}]},
}
+ updated_crawler_description = "Glue crawler updated by the Amazon provider
system test"
+ updated_glue_crawler_config = {
+ **glue_crawler_config,
+ "Description": updated_crawler_description,
+ }
create_bucket = S3CreateBucketOperator(
task_id="create_bucket",
@@ -127,15 +145,37 @@ with DAG(
replace=True,
)
- # [START howto_operator_glue_crawler]
- crawl_s3 = GlueCrawlerOperator(
- task_id="crawl_s3",
+ # [START howto_operator_glue_crawler_create]
+ create_crawler = GlueCrawlerCreateOperator(
+ task_id="create_crawler",
config=glue_crawler_config,
)
- # [END howto_operator_glue_crawler]
+ # [END howto_operator_glue_crawler_create]
+
+ # [START howto_operator_glue_crawler_update]
+ update_crawler = GlueCrawlerUpdateOperator(
+ task_id="update_crawler",
+ config=updated_glue_crawler_config,
+ )
+ # [END howto_operator_glue_crawler_update]
+
+ verify_crawler_update = verify_crawler_description(glue_crawler_name,
updated_crawler_description)
+
+ # [START howto_operator_glue_crawler_run_deferrable]
+ run_crawler_deferrable = GlueCrawlerRunOperator(
+ task_id="run_crawler_deferrable",
+ crawler_name=glue_crawler_name,
+ deferrable=True,
+ )
+ # [END howto_operator_glue_crawler_run_deferrable]
- # GlueCrawlerOperator waits by default, setting as False to test the
Sensor below.
- crawl_s3.wait_for_completion = False
+ # [START howto_operator_glue_crawler_run]
+ run_crawler = GlueCrawlerRunOperator(
+ task_id="run_crawler",
+ crawler_name=glue_crawler_name,
+ deferrable=False,
+ )
+ # [END howto_operator_glue_crawler_run]
# [START howto_sensor_glue_crawler]
wait_for_crawl = GlueCrawlerSensor(
@@ -179,6 +219,14 @@ with DAG(
# [END howto_sensor_glue]
wait_for_job.poke_interval = 5
+ # [START howto_operator_glue_crawler_delete]
+ delete_crawler = GlueCrawlerDeleteOperator(
+ task_id="delete_crawler",
+ crawler_name=glue_crawler_name,
+ trigger_rule=TriggerRule.ALL_DONE,
+ )
+ # [END howto_operator_glue_crawler_delete]
+
delete_bucket = S3DeleteBucketOperator(
task_id="delete_bucket",
trigger_rule=TriggerRule.ALL_DONE,
@@ -204,13 +252,18 @@ with DAG(
upload_csv,
upload_script,
# TEST BODY
- crawl_s3,
+ create_crawler,
+ update_crawler,
+ verify_crawler_update,
+ run_crawler_deferrable,
+ run_crawler,
wait_for_crawl,
wait_for_catalog_partition,
submit_glue_job,
wait_for_job,
# TEST TEARDOWN
- glue_cleanup(glue_crawler_name, glue_job_name, glue_db_name),
+ delete_crawler,
+ delete_glue_resources(glue_job_name, glue_db_name),
delete_bucket,
log_cleanup,
)
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_glue_crawler.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_glue_crawler.py
index f56c4d30b77..ff584881917 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_glue_crawler.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_glue_crawler.py
@@ -24,9 +24,17 @@ import pytest
from botocore.exceptions import ClientError
from moto import mock_aws
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.amazon.aws.hooks.glue_crawler import GlueCrawlerHook
from airflow.providers.amazon.aws.hooks.sts import StsHook
-from airflow.providers.amazon.aws.operators.glue_crawler import
GlueCrawlerOperator
+from airflow.providers.amazon.aws.operators.glue_crawler import (
+ GlueCrawlerCreateOperator,
+ GlueCrawlerDeleteOperator,
+ GlueCrawlerOperator,
+ GlueCrawlerRunOperator,
+ GlueCrawlerUpdateOperator,
+)
+from airflow.providers.common.compat.sdk import AirflowException
from unit.amazon.aws.utils.test_template_fields import validate_template_fields
@@ -93,6 +101,164 @@ mock_config = {
}
+class TestGlueCrawlerCreateOperator:
+ @mock.patch.object(GlueCrawlerHook, "create_crawler", autospec=True)
+ def test_execute(self, mock_create_crawler):
+ op = GlueCrawlerCreateOperator(task_id="create_crawler",
config=mock_config)
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_create_crawler.assert_called_once_with(op.hook, **mock_config)
+
+ def test_template_fields(self):
+ op = GlueCrawlerCreateOperator(task_id="create_crawler",
config=mock_config)
+
+ validate_template_fields(op)
+
+
+class TestGlueCrawlerUpdateOperator:
+ @mock.patch.object(GlueCrawlerHook, "update_crawler", autospec=True)
+ def test_execute(self, mock_update_crawler):
+ op = GlueCrawlerUpdateOperator(task_id="update_crawler",
config=mock_config)
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_update_crawler.assert_called_once_with(op.hook, **mock_config)
+
+ def test_template_fields(self):
+ op = GlueCrawlerUpdateOperator(task_id="update_crawler",
config=mock_config)
+
+ validate_template_fields(op)
+
+
+class TestGlueCrawlerRunOperator:
+ @mock.patch.object(GlueCrawlerHook, "wait_for_crawler_completion",
autospec=True)
+ @mock.patch.object(GlueCrawlerHook, "start_crawler", autospec=True)
+ def test_execute_waits_for_completion_by_default(self, mock_start_crawler,
mock_wait):
+ op = GlueCrawlerRunOperator(task_id="run_crawler",
crawler_name=mock_crawler_name)
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_start_crawler.assert_called_once_with(op.hook, mock_crawler_name)
+ mock_wait.assert_called_once_with(op.hook,
crawler_name=mock_crawler_name, poll_interval=5)
+
+ @mock.patch.object(GlueCrawlerHook, "wait_for_crawler_completion",
autospec=True)
+ @mock.patch.object(GlueCrawlerHook, "start_crawler", autospec=True)
+ def test_execute_without_waiting(self, mock_start_crawler, mock_wait):
+ op = GlueCrawlerRunOperator(
+ task_id="run_crawler", crawler_name=mock_crawler_name,
wait_for_completion=False
+ )
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_start_crawler.assert_called_once_with(op.hook, mock_crawler_name)
+ mock_wait.assert_not_called()
+
+ @mock.patch.object(GlueCrawlerHook, "wait_for_crawler_completion",
autospec=True)
+ @mock.patch.object(GlueCrawlerHook, "start_crawler", autospec=True)
+ def test_execute_defers(self, mock_start_crawler, mock_wait):
+ op = GlueCrawlerRunOperator(
+ task_id="run_crawler",
+ crawler_name=mock_crawler_name,
+ poll_interval=10,
+ deferrable=True,
+ aws_conn_id="fake-conn-id",
+ region_name="eu-west-2",
+ verify=False,
+ botocore_config={"read_timeout": 42},
+ )
+ op.defer = mock.MagicMock(spec=op.defer)
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_start_crawler.assert_called_once_with(op.hook, mock_crawler_name)
+ mock_wait.assert_not_called()
+ op.defer.assert_called_once()
+ trigger = op.defer.call_args.kwargs["trigger"]
+ _, trigger_kwargs = trigger.serialize()
+ assert trigger_kwargs == {
+ "crawler_name": mock_crawler_name,
+ "waiter_delay": 10,
+ "waiter_max_attempts": 1500,
+ "aws_conn_id": "fake-conn-id",
+ "region_name": "eu-west-2",
+ "verify": False,
+ "botocore_config": {"read_timeout": 42},
+ }
+ assert op.defer.call_args.kwargs["method_name"] == "execute_complete"
+
+ @mock.patch.object(GlueCrawlerHook, "wait_for_crawler_completion",
autospec=True)
+ @mock.patch.object(GlueCrawlerHook, "start_crawler", autospec=True)
+ def test_execute_waits_when_crawler_is_already_running(self,
mock_start_crawler, mock_wait):
+ mock_start_crawler.side_effect = ClientError(
+ error_response={"Error": {"Code": "CrawlerRunningException",
"Message": "Already running"}},
+ operation_name="StartCrawler",
+ )
+ op = GlueCrawlerRunOperator(
+ task_id="run_crawler",
+ crawler_name=mock_crawler_name,
+ fail_on_already_running=False,
+ )
+
+ result = op.execute({})
+
+ assert result == mock_crawler_name
+ mock_wait.assert_called_once_with(op.hook,
crawler_name=mock_crawler_name, poll_interval=5)
+
+ @pytest.mark.parametrize("error_code", ["CrawlerRunningException",
"EntityNotFoundException"])
+ @mock.patch.object(GlueCrawlerHook, "start_crawler", autospec=True)
+ def test_execute_propagates_client_error(self, mock_start_crawler,
error_code):
+ mock_start_crawler.side_effect = ClientError(
+ error_response={"Error": {"Code": error_code, "Message": "error"}},
+ operation_name="StartCrawler",
+ )
+ op = GlueCrawlerRunOperator(task_id="run_crawler",
crawler_name=mock_crawler_name)
+
+ with pytest.raises(ClientError) as exc_info:
+ op.execute({})
+
+ assert exc_info.value.response["Error"]["Code"] == error_code
+
+ def test_execute_complete(self):
+ op = GlueCrawlerRunOperator(task_id="run_crawler",
crawler_name=mock_crawler_name)
+
+ assert op.execute_complete({}, {"status": "success"}) ==
mock_crawler_name
+
+ def test_execute_complete_raises_for_failure(self):
+ op = GlueCrawlerRunOperator(task_id="run_crawler",
crawler_name=mock_crawler_name)
+
+ with pytest.raises(AirflowException, match="Error in glue crawl"):
+ op.execute_complete({}, {"status": "error"})
+
+ def test_template_fields(self):
+ op = GlueCrawlerRunOperator(task_id="run_crawler",
crawler_name=mock_crawler_name)
+
+ validate_template_fields(op)
+
+
+class TestGlueCrawlerDeleteOperator:
+ @mock.patch.object(GlueCrawlerHook, "glue_client",
new_callable=mock.PropertyMock)
+ def test_execute(self, mock_glue_client):
+ client = mock.MagicMock(spec=["delete_crawler"])
+ mock_glue_client.return_value = client
+ op = GlueCrawlerDeleteOperator(task_id="delete_crawler",
crawler_name=mock_crawler_name)
+
+ result = op.execute({})
+
+ assert result is None
+ client.delete_crawler.assert_called_once_with(Name=mock_crawler_name)
+
+ def test_template_fields(self):
+ op = GlueCrawlerDeleteOperator(task_id="delete_crawler",
crawler_name=mock_crawler_name)
+
+ validate_template_fields(op)
+
+
class TestGlueCrawlerOperator:
@pytest.fixture
def mock_conn(self) -> Generator[BaseAwsConnection, None, None]:
@@ -107,17 +273,19 @@ class TestGlueCrawlerOperator:
yield hook
def setup_method(self):
- self.op = GlueCrawlerOperator(task_id="test_glue_crawler_operator",
config=mock_config)
+ with pytest.warns(AirflowProviderDeprecationWarning,
match="GlueCrawlerOperator is deprecated"):
+ self.op =
GlueCrawlerOperator(task_id="test_glue_crawler_operator", config=mock_config)
def test_init(self):
- op = GlueCrawlerOperator(
- task_id="test_glue_crawler_operator",
- aws_conn_id="fake-conn-id",
- region_name="eu-west-2",
- verify=True,
- botocore_config={"read_timeout": 42},
- config=mock_config,
- )
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ op = GlueCrawlerOperator(
+ task_id="test_glue_crawler_operator",
+ aws_conn_id="fake-conn-id",
+ region_name="eu-west-2",
+ verify=True,
+ botocore_config={"read_timeout": 42},
+ config=mock_config,
+ )
assert op.hook.client_type == "glue"
assert op.hook.resource_type is None
@@ -127,7 +295,8 @@ class TestGlueCrawlerOperator:
assert op.hook._config is not None
assert op.hook._config.read_timeout == 42
- op = GlueCrawlerOperator(task_id="test_glue_crawler_operator",
config=mock_config)
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ op = GlueCrawlerOperator(task_id="test_glue_crawler_operator",
config=mock_config)
assert op.hook.aws_conn_id == "aws_default"
assert op.hook._region_name is None
@@ -181,6 +350,13 @@ class TestGlueCrawlerOperator:
def test_template_fields(self):
validate_template_fields(self.op)
+ def test_execute_complete_returns_name_from_config(self):
+ self.op.crawler_name = "unrendered-name"
+
+ result = self.op.execute_complete({}, {"status": "success"})
+
+ assert result == mock_config["Name"]
+
@mock.patch.object(GlueCrawlerHook, "wait_for_crawler_completion")
@mock.patch.object(GlueCrawlerHook, "start_crawler")
@mock.patch.object(GlueCrawlerHook, "update_crawler")
