This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 7503189ecd1 Add `GlueCatalogCreateDatabaseOperator` (#65977)
7503189ecd1 is described below
commit 7503189ecd1ad590d62e02a42b33e1aa66e2a444
Author: John Jackson <[email protected]>
AuthorDate: Tue Apr 28 06:56:10 2026 -0700
Add `GlueCatalogCreateDatabaseOperator` (#65977)
---
providers/amazon/docs/operators/glue_catalog.rst | 42 ++++++++
providers/amazon/provider.yaml | 2 +
.../providers/amazon/aws/operators/glue_catalog.py | 109 +++++++++++++++++++++
.../airflow/providers/amazon/get_provider_info.py | 6 +-
.../system/amazon/aws/example_glue_catalog.py | 74 ++++++++++++++
.../unit/amazon/aws/operators/test_glue_catalog.py | 93 ++++++++++++++++++
6 files changed, 325 insertions(+), 1 deletion(-)
diff --git a/providers/amazon/docs/operators/glue_catalog.rst
b/providers/amazon/docs/operators/glue_catalog.rst
new file mode 100644
index 00000000000..f42be0fc67e
--- /dev/null
+++ b/providers/amazon/docs/operators/glue_catalog.rst
@@ -0,0 +1,42 @@
+ .. 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.
+
+======================
+AWS Glue Data Catalog
+======================
+
+The AWS Glue Data Catalog is a centralized metadata repository for data assets.
+Use the operators below to manage Glue Data Catalog resources.
+
+.. _howto/operator:GlueCatalogCreateDatabaseOperator:
+
+Create a Catalog Database
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+To create a database in the AWS Glue Data Catalog, use
+:class:`~airflow.providers.amazon.aws.operators.glue_catalog.GlueCatalogCreateDatabaseOperator`.
+
+.. exampleinclude::
/../../amazon/tests/system/amazon/aws/example_glue_catalog.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_glue_catalog_create_database]
+ :end-before: [END howto_operator_glue_catalog_create_database]
+
+Reference
+~~~~~~~~~
+
+* `AWS boto3 Library Documentation for Glue
<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/glue.html>`__
diff --git a/providers/amazon/provider.yaml b/providers/amazon/provider.yaml
index 10bef30b640..1306418c31e 100644
--- a/providers/amazon/provider.yaml
+++ b/providers/amazon/provider.yaml
@@ -330,6 +330,7 @@ integrations:
logo: /docs/integration-logos/[email protected]
how-to-guide:
- /docs/apache-airflow-providers-amazon/operators/glue.rst
+ - /docs/apache-airflow-providers-amazon/operators/glue_catalog.rst
tags: [aws]
- integration-name: AWS Lambda
external-doc-url: https://aws.amazon.com/lambda/
@@ -444,6 +445,7 @@ operators:
- integration-name: AWS Glue
python-modules:
- airflow.providers.amazon.aws.operators.glue
+ - airflow.providers.amazon.aws.operators.glue_catalog
- airflow.providers.amazon.aws.operators.glue_crawler
- integration-name: AWS Lambda
python-modules:
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_catalog.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_catalog.py
new file mode 100644
index 00000000000..7ed9732c386
--- /dev/null
+++
b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue_catalog.py
@@ -0,0 +1,109 @@
+# 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.
+"""AWS Glue Data Catalog operators."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Literal
+
+from botocore.exceptions import ClientError
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
+from airflow.utils.helpers import prune_dict
+
+if TYPE_CHECKING:
+ from airflow.sdk import Context
+
+
+class GlueCatalogCreateDatabaseOperator(AwsBaseOperator[AwsBaseHook]):
+ """
+ Create a database in the AWS Glue Data Catalog.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:GlueCatalogCreateDatabaseOperator`
+
+ :param database_name: The name of the database to create.
+ :param description: A description of the database.
+ :param location_uri: The location of the database (e.g. an S3 path).
+ :param parameters: Key-value pairs that define properties of the database.
+ :param catalog_id: The ID of the Data Catalog. Defaults to the account ID.
+ :param tags: Tags to assign to the database.
+ :param if_exists: Behavior when the database already exists.
+ ``"fail"`` raises an error, ``"skip"`` logs and returns the database
name.
+ """
+
+ aws_hook_class = AwsBaseHook
+ template_fields: tuple[str, ...] = (
+ *AwsBaseOperator.template_fields,
+ "database_name",
+ "description",
+ "location_uri",
+ )
+
+ def __init__(
+ self,
+ *,
+ database_name: str,
+ description: str | None = None,
+ location_uri: str | None = None,
+ parameters: dict[str, str] | None = None,
+ catalog_id: str | None = None,
+ tags: dict[str, str] | None = None,
+ if_exists: Literal["fail", "skip"] = "skip",
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.database_name = database_name
+ self.description = description
+ self.location_uri = location_uri
+ self.parameters = parameters
+ self.catalog_id = catalog_id
+ self.tags = tags
+ self.if_exists = if_exists
+
+ @property
+ def _hook_parameters(self) -> dict[str, Any]:
+ return {**super()._hook_parameters, "client_type": "glue"}
+
+ def execute(self, context: Context) -> str:
+ database_input: dict[str, Any] = prune_dict(
+ {
+ "Name": self.database_name,
+ "Description": self.description,
+ "LocationUri": self.location_uri,
+ "Parameters": self.parameters,
+ }
+ )
+ kwargs: dict[str, Any] = prune_dict(
+ {
+ "DatabaseInput": database_input,
+ "CatalogId": self.catalog_id,
+ "Tags": self.tags,
+ }
+ )
+ try:
+ self.hook.conn.create_database(**kwargs)
+ except ClientError as e:
+ if e.response["Error"]["Code"] == "AlreadyExistsException" and
self.if_exists == "skip":
+ self.log.info("Database %s already exists, skipping.",
self.database_name)
+ else:
+ raise
+ else:
+ self.log.info("Created Glue Catalog database: %s",
self.database_name)
+ return self.database_name
diff --git a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
index aac6b05f3c7..4b341628140 100644
--- a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
+++ b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
@@ -284,7 +284,10 @@ def get_provider_info():
"integration-name": "AWS Glue",
"external-doc-url": "https://aws.amazon.com/glue/";,
"logo": "/docs/integration-logos/[email protected]",
- "how-to-guide":
["/docs/apache-airflow-providers-amazon/operators/glue.rst"],
+ "how-to-guide": [
+ "/docs/apache-airflow-providers-amazon/operators/glue.rst",
+
"/docs/apache-airflow-providers-amazon/operators/glue_catalog.rst",
+ ],
"tags": ["aws"],
},
{
@@ -428,6 +431,7 @@ def get_provider_info():
"integration-name": "AWS Glue",
"python-modules": [
"airflow.providers.amazon.aws.operators.glue",
+ "airflow.providers.amazon.aws.operators.glue_catalog",
"airflow.providers.amazon.aws.operators.glue_crawler",
],
},
diff --git a/providers/amazon/tests/system/amazon/aws/example_glue_catalog.py
b/providers/amazon/tests/system/amazon/aws/example_glue_catalog.py
new file mode 100644
index 00000000000..80246706dfb
--- /dev/null
+++ b/providers/amazon/tests/system/amazon/aws/example_glue_catalog.py
@@ -0,0 +1,74 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.providers.amazon.aws.operators.glue_catalog import
GlueCatalogCreateDatabaseOperator
+from airflow.providers.common.compat.sdk import DAG, chain
+
+from system.amazon.aws.utils import ENV_ID_KEY, SystemTestContextBuilder
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+ from airflow.sdk import TriggerRule, task
+else:
+ from airflow.decorators import task # type: ignore[attr-defined,no-redef]
+ from airflow.utils.trigger_rule import TriggerRule # type:
ignore[no-redef,attr-defined]
+
+DAG_ID = "example_glue_catalog"
+
+sys_test_context_task = SystemTestContextBuilder().build()
+
+with DAG(
+ dag_id=DAG_ID,
+ schedule=None,
+ start_date=datetime(2024, 1, 1),
+ catchup=False,
+) as dag:
+ test_context = sys_test_context_task()
+ env_id = test_context[ENV_ID_KEY]
+ db_name = f"{env_id}_test_db"
+
+ # [START howto_operator_glue_catalog_create_database]
+ create_database = GlueCatalogCreateDatabaseOperator(
+ task_id="create_database",
+ database_name=db_name,
+ description="Test database for Glue Catalog",
+ )
+ # [END howto_operator_glue_catalog_create_database]
+
+ @task(trigger_rule=TriggerRule.ALL_DONE)
+ def delete_database(name: str):
+ """Delete the Glue Catalog database."""
+ import boto3
+
+ boto3.client("glue").delete_database(Name=name)
+
+ chain(
+ test_context,
+ create_database,
+ delete_database(name=db_name),
+ )
+
+ from tests_common.test_utils.watcher import watcher
+
+ list(dag.tasks) >> watcher()
+
+from tests_common.test_utils.system_tests import get_test_run # noqa: E402
+
+test_run = get_test_run(dag)
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_glue_catalog.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_glue_catalog.py
new file mode 100644
index 00000000000..10b4cfda919
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_glue_catalog.py
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from botocore.exceptions import ClientError
+
+from airflow.providers.amazon.aws.operators.glue_catalog import
GlueCatalogCreateDatabaseOperator
+
+from unit.amazon.aws.utils.test_template_fields import validate_template_fields
+
+DB_NAME = "test_database"
+
+
+class TestGlueCatalogCreateDatabaseOperator:
+ def test_execute(self):
+ op = GlueCatalogCreateDatabaseOperator(
+ task_id="create_db",
+ database_name=DB_NAME,
+ )
+ mock_conn = MagicMock()
+ op.hook.conn = mock_conn
+
+ result = op.execute({})
+
+
mock_conn.create_database.assert_called_once_with(DatabaseInput={"Name":
DB_NAME})
+ assert result == DB_NAME
+
+ def test_execute_with_all_params(self):
+ op = GlueCatalogCreateDatabaseOperator(
+ task_id="create_db",
+ database_name=DB_NAME,
+ description="Test database",
+ location_uri="s3://my-bucket/db/",
+ parameters={"key": "value"},
+ catalog_id="123456789012",
+ tags={"env": "test"},
+ )
+ mock_conn = MagicMock()
+ op.hook.conn = mock_conn
+
+ result = op.execute({})
+
+ mock_conn.create_database.assert_called_once_with(
+ DatabaseInput={
+ "Name": DB_NAME,
+ "Description": "Test database",
+ "LocationUri": "s3://my-bucket/db/",
+ "Parameters": {"key": "value"},
+ },
+ CatalogId="123456789012",
+ Tags={"env": "test"},
+ )
+ assert result == DB_NAME
+
+ def test_execute_skip_existing(self):
+ op = GlueCatalogCreateDatabaseOperator(
+ task_id="create_db",
+ database_name=DB_NAME,
+ if_exists="skip",
+ )
+ mock_conn = MagicMock()
+ mock_conn.create_database.side_effect = ClientError(
+ {"Error": {"Code": "AlreadyExistsException", "Message": "Already
exists"}},
+ "CreateDatabase",
+ )
+ op.hook.conn = mock_conn
+
+ result = op.execute({})
+
+ assert result == DB_NAME
+
+ def test_template_fields(self):
+ op = GlueCatalogCreateDatabaseOperator(
+ task_id="test",
+ database_name=DB_NAME,
+ )
+ validate_template_fields(op)
