[AIRFLOW-6689] Last changes for AIP-21

airflow/operators/presto_to_mysql.py

@@ -16,65 +16,14 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Optional
+"""This module is deprecated. Please use `airflow.providers.mysql.operators.presto_to_mysql`."""
 
-from airflow.models import BaseOperator
-from airflow.providers.mysql.hooks.mysql import MySqlHook
-from airflow.providers.presto.hooks.presto import PrestoHook
-from airflow.utils.decorators import apply_defaults
+import warnings
 
+# pylint: disable=unused-import
+from airflow.providers.mysql.operators.presto_to_mysql import PrestoToMySqlTransfer  # noqa
 
-class PrestoToMySqlTransfer(BaseOperator):
-    """
-    Moves data from Presto to MySQL, note that for now the data is loaded
-    into memory before being pushed to MySQL, so this operator should
-    be used for smallish amount of data.
-
-    :param sql: SQL query to execute against Presto. (templated)
-    :type sql: str
-    :param mysql_table: target MySQL table, use dot notation to target a
-        specific database. (templated)
-    :type mysql_table: str
-    :param mysql_conn_id: source mysql connection
-    :type mysql_conn_id: str
-    :param presto_conn_id: source presto connection
-    :type presto_conn_id: str
-    :param mysql_preoperator: sql statement to run against mysql prior to
-        import, typically use to truncate of delete in place
-        of the data coming in, allowing the task to be idempotent (running
-        the task twice won't double load data). (templated)
-    :type mysql_preoperator: str
-    """
-
-    template_fields = ('sql', 'mysql_table', 'mysql_preoperator')
-    template_ext = ('.sql',)
-    ui_color = '#a0e08c'
-
-    @apply_defaults
-    def __init__(self,
-                 sql: str,
-                 mysql_table: str,
-                 presto_conn_id: str = 'presto_default',
-                 mysql_conn_id: str = 'mysql_default',
-                 mysql_preoperator: Optional[str] = None,
-                 *args, **kwargs) -> None:
-        super().__init__(*args, **kwargs)
-        self.sql = sql
-        self.mysql_table = mysql_table
-        self.mysql_conn_id = mysql_conn_id
-        self.mysql_preoperator = mysql_preoperator
-        self.presto_conn_id = presto_conn_id
-
-    def execute(self, context):
-        presto = PrestoHook(presto_conn_id=self.presto_conn_id)
-        self.log.info("Extracting data from Presto: %s", self.sql)
-        results = presto.get_records(self.sql)
-
-        mysql = MySqlHook(mysql_conn_id=self.mysql_conn_id)
-        if self.mysql_preoperator:
-            self.log.info("Running MySQL preoperator")
-            self.log.info(self.mysql_preoperator)
-            mysql.run(self.mysql_preoperator)
-
-        self.log.info("Inserting rows into MySQL")
-        mysql.insert_rows(table=self.mysql_table, rows=results)
+warnings.warn(
+    "This module is deprecated. Please use `airflow.providers.mysql.operators.presto_to_mysql`.",
+    DeprecationWarning, stacklevel=2
+)

airflow/providers/mysql/operators/presto_to_mysql.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+#
+# 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 Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.utils.decorators import apply_defaults
+
+
+class PrestoToMySqlTransfer(BaseOperator):
+    """
+    Moves data from Presto to MySQL, note that for now the data is loaded
+    into memory before being pushed to MySQL, so this operator should
+    be used for smallish amount of data.
+
+    :param sql: SQL query to execute against Presto. (templated)
+    :type sql: str
+    :param mysql_table: target MySQL table, use dot notation to target a
+        specific database. (templated)
+    :type mysql_table: str
+    :param mysql_conn_id: source mysql connection
+    :type mysql_conn_id: str
+    :param presto_conn_id: source presto connection
+    :type presto_conn_id: str
+    :param mysql_preoperator: sql statement to run against mysql prior to
+        import, typically use to truncate of delete in place
+        of the data coming in, allowing the task to be idempotent (running
+        the task twice won't double load data). (templated)
+    :type mysql_preoperator: str
+    """
+
+    template_fields = ('sql', 'mysql_table', 'mysql_preoperator')
+    template_ext = ('.sql',)
+    ui_color = '#a0e08c'
+
+    @apply_defaults
+    def __init__(self,
+                 sql: str,
+                 mysql_table: str,
+                 presto_conn_id: str = 'presto_default',
+                 mysql_conn_id: str = 'mysql_default',
+                 mysql_preoperator: Optional[str] = None,
+                 *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.sql = sql
+        self.mysql_table = mysql_table
+        self.mysql_conn_id = mysql_conn_id
+        self.mysql_preoperator = mysql_preoperator
+        self.presto_conn_id = presto_conn_id
+
+    def execute(self, context):
+        presto = PrestoHook(presto_conn_id=self.presto_conn_id)
+        self.log.info("Extracting data from Presto: %s", self.sql)
+        results = presto.get_records(self.sql)
+
+        mysql = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+        if self.mysql_preoperator:
+            self.log.info("Running MySQL preoperator")
+            self.log.info(self.mysql_preoperator)
+            mysql.run(self.mysql_preoperator)
+
+        self.log.info("Inserting rows into MySQL")
+        mysql.insert_rows(table=self.mysql_table, rows=results)

docs/concepts.rst

@@ -278,7 +278,7 @@ Airflow provides operators for many common tasks, including:
 In addition to these basic building blocks, there are many more specific
 operators: :class:`~airflow.providers.docker.operators.docker.DockerOperator`,
 :class:`~airflow.providers.apache.hive.operators.hive.HiveOperator`, :class:`~airflow.providers.amazon.aws.operators.s3_file_transform.S3FileTransformOperator`,
-:class:`~airflow.operators.presto_to_mysql.PrestoToMySqlTransfer`,
+:class:`~airflow.providers.mysql.operators.presto_to_mysql.PrestoToMySqlTransfer`,
 :class:`~airflow.providers.slack.operators.slack.SlackAPIOperator`... you get the idea!
 
 Operators are only loaded by Airflow if they are assigned to a DAG.

docs/operators-and-hooks-ref.rst

@@ -1213,7 +1213,7 @@ These integrations allow you to copy data.
    * - `Presto <https://prestodb.github.io/>`__
      - `MySQL <https://www.mysql.com/>`__
      -
-     - :mod:`airflow.operators.presto_to_mysql`
+     - :mod:`airflow.providers.mysql.operators.presto_to_mysql`
 
    * - SQL
      - `Cloud Storage (GCS) <https://cloud.google.com/gcs/>`__

scripts/ci/pylint_todo.txt

@@ -156,7 +156,7 @@
 ./airflow/providers/apache/pig/operators/pig.py
 ./airflow/providers/postgres/operators/postgres.py
 ./airflow/providers/presto/operators/presto_check.py
-./airflow/operators/presto_to_mysql.py
+./airflow/providers/mysql/operators/presto_to_mysql.py
 ./airflow/operators/python.py
 ./airflow/providers/amazon/aws/operators/s3_file_transform.py
 ./airflow/providers/amazon/aws/operators/s3_to_redshift.py