From 88fe2f0783861870d8603fd489caa5901a70bbbf Mon Sep 17 00:00:00 2001 From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com> Date: Wed, 15 Jul 2026 12:48:31 -0300 Subject: [PATCH 1/4] Add dedicated Glue crawler lifecycle operators 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. --- providers/amazon/docs/operators/glue.rst | 68 ++++- .../amazon/aws/operators/glue_crawler.py | 254 ++++++++++++++---- .../tests/system/amazon/aws/example_glue.py | 75 +++++- .../amazon/aws/operators/test_glue_crawler.py | 198 +++++++++++++- 4 files changed, 523 insertions(+), 72 deletions(-) diff --git a/providers/amazon/docs/operators/glue.rst b/providers/amazon/docs/operators/glue.rst index c3e156c4dc8ea..e90ddeae24c30 100644 --- a/providers/amazon/docs/operators/glue.rst +++ b/providers/amazon/docs/operators/glue.rst @@ -37,26 +37,84 @@ 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: + +.. 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. + .. _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 7c8e0d54d5bac..917875372cd84 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 @@ 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 @@ def __init__( 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 @@ def execute(self, context: Context) -> str: 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,128 @@ def execute(self, context: Context) -> str: ) 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) + + +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 d62bcf3b30883..b6881a5683e0a 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 @@ 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 @@ @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 @@ def glue_cleanup(crawler_name: str, job_name: str, db_name: str) -> None: 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 @@ def glue_cleanup(crawler_name: str, job_name: str, db_name: str) -> None: 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 @@ def glue_cleanup(crawler_name: str, job_name: str, db_name: str) -> None: # [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 @@ def glue_cleanup(crawler_name: str, job_name: str, db_name: str) -> None: 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 f56c4d30b772c..ff58488191739 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 @@ 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 @@ } +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 @@ def crawler_hook(self) -> Generator[GlueCrawlerHook, None, None]: 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 @@ def test_init(self): 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 @@ def test_crawler_wait_combinations(self, _, wait_for_completion, deferrable, moc 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") From 78a848b2cbfaa3fdd391967abb5b7c37949c64bd Mon Sep 17 00:00:00 2001 From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com> Date: Wed, 15 Jul 2026 13:08:38 -0300 Subject: [PATCH 2/4] Keep deprecated Glue crawler operator documented Existing users and external documentation links need a clear description of the legacy behavior while Dags migrate to operation-specific tasks. --- providers/amazon/docs/operators/glue.rst | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/providers/amazon/docs/operators/glue.rst b/providers/amazon/docs/operators/glue.rst index e90ddeae24c30..e4a9466ba2266 100644 --- a/providers/amazon/docs/operators/glue.rst +++ b/providers/amazon/docs/operators/glue.rst @@ -110,11 +110,24 @@ To delete an existing crawler, use .. _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 From 3ab00f37c752a388b480c40fd061e3a4468c9ac0 Mon Sep 17 00:00:00 2001 From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:02:43 -0300 Subject: [PATCH 3/4] Register Glue crawler classes in example checks Project structure validation needs to distinguish the internal base class and the intentionally deprecated public operator from classes requiring new examples. --- airflow-core/tests/unit/always/test_project_structure.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/airflow-core/tests/unit/always/test_project_structure.py b/airflow-core/tests/unit/always/test_project_structure.py index 750966a320221..9286bdb5e41a5 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", } From aaea279dacfc11ab7a880731ca8dbe7a2a7fc498 Mon Sep 17 00:00:00 2001 From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com> Date: Thu, 16 Jul 2026 15:18:41 -0300 Subject: [PATCH 4/4] Document Glue crawler operator removal window Users need at least one full major provider release cycle to migrate from this widely used operator. --- .../src/airflow/providers/amazon/aws/operators/glue_crawler.py | 1 + 1 file changed, 1 insertion(+) 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 917875372cd84..29ad429276f07 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 @@ -240,6 +240,7 @@ def execute(self, context: Context) -> None: 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.