#
# 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.
# pylint: disable=too-many-lines
"""This module contains a Google Cloud SQL Hook."""

import errno
import json
import os
import os.path
import platform
import random
import re
import shutil
import socket
import string
import subprocess
import time
import uuid
from pathlib import Path
from subprocess import PIPE, Popen
from typing import Any, Dict, List, Optional, Sequence, Union
from urllib.parse import quote_plus

import requests
from googleapiclient.discovery import Resource, build
from googleapiclient.errors import HttpError
from sqlalchemy.orm import Session

from airflow.exceptions import AirflowException

# Number of retries - used by googleapiclient method calls to perform retries
# For requests that are "retriable"
from airflow.hooks.base import BaseHook
from airflow.models import Connection
from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.session import provide_session

UNIX_PATH_MAX = 108

# Time to sleep between active checks of the operation results
TIME_TO_SLEEP_IN_SECONDS = 20


class CloudSqlOperationStatus:
    """Helper class with operation statuses."""

    PENDING = "PENDING"
    RUNNING = "RUNNING"
    DONE = "DONE"
    UNKNOWN = "UNKNOWN"


class CloudSQLHook(GoogleBaseHook):
    """
    Hook for Google Cloud SQL APIs.

    All the methods in the hook where project_id is used must be called with
    keyword arguments rather than positional.
    """

    conn_name_attr = 'gcp_conn_id'
    default_conn_name = 'google_cloud_default'
    conn_type = 'gcpcloudsql'
    hook_name = 'Google Cloud SQL'

    def __init__(
        self,
        api_version: str,
        gcp_conn_id: str = default_conn_name,
        delegate_to: Optional[str] = None,
        impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
    ) -> None:
        super().__init__(
            gcp_conn_id=gcp_conn_id,
            delegate_to=delegate_to,
            impersonation_chain=impersonation_chain,
        )
        self.api_version = api_version
        self._conn = None

    def get_conn(self) -> Resource:
        """
        Retrieves connection to Cloud SQL.

        :return: Google Cloud SQL services object.
        :rtype: dict
        """
        if not self._conn:
            http_authorized = self._authorize()
            self._conn = build('sqladmin', self.api_version, http=http_authorized, cache_discovery=False)
        return self._conn

    @GoogleBaseHook.fallback_to_default_project_id
    def get_instance(self, instance: str, project_id: str) -> dict:
        """
        Retrieves a resource containing information about a Cloud SQL instance.

        :param instance: Database instance ID. This does not include the project ID.
        :type instance: str
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: A Cloud SQL instance resource.
        :rtype: dict
        """
        return (
            self.get_conn()  # noqa # pylint: disable=no-member
            .instances()
            .get(project=project_id, instance=instance)
            .execute(num_retries=self.num_retries)
        )

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def create_instance(self, body: Dict, project_id: str) -> None:
        """
        Creates a new Cloud SQL instance.

        :param body: Body required by the Cloud SQL insert API, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/insert#request-body.
        :type body: dict
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .instances()
            .insert(project=project_id, body=body)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def patch_instance(self, body: dict, instance: str, project_id: str) -> None:
        """
        Updates settings of a Cloud SQL instance.

        Caution: This is not a partial update, so you must include values for
        all the settings that you want to retain.

        :param body: Body required by the Cloud SQL patch API, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/patch#request-body.
        :type body: dict
        :param instance: Cloud SQL instance ID. This does not include the project ID.
        :type instance: str
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .instances()
            .patch(project=project_id, instance=instance, body=body)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def delete_instance(self, instance: str, project_id: str) -> None:
        """
        Deletes a Cloud SQL instance.

        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :param instance: Cloud SQL instance ID. This does not include the project ID.
        :type instance: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .instances()
            .delete(project=project_id, instance=instance)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    def get_database(self, instance: str, database: str, project_id: str) -> dict:
        """
        Retrieves a database resource from a Cloud SQL instance.

        :param instance: Database instance ID. This does not include the project ID.
        :type instance: str
        :param database: Name of the database in the instance.
        :type database: str
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: A Cloud SQL database resource, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases#resource.
        :rtype: dict
        """
        return (
            self.get_conn()  # noqa # pylint: disable=no-member
            .databases()
            .get(project=project_id, instance=instance, database=database)
            .execute(num_retries=self.num_retries)
        )

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def create_database(self, instance: str, body: Dict, project_id: str) -> None:
        """
        Creates a new database inside a Cloud SQL instance.

        :param instance: Database instance ID. This does not include the project ID.
        :type instance: str
        :param body: The request body, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases/insert#request-body.
        :type body: dict
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .databases()
            .insert(project=project_id, instance=instance, body=body)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def patch_database(
        self,
        instance: str,
        database: str,
        body: Dict,
        project_id: str,
    ) -> None:
        """
        Updates a database resource inside a Cloud SQL instance.

        This method supports patch semantics.
        See https://cloud.google.com/sql/docs/mysql/admin-api/how-tos/performance#patch.

        :param instance: Database instance ID. This does not include the project ID.
        :type instance: str
        :param database: Name of the database to be updated in the instance.
        :type database: str
        :param body: The request body, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases/insert#request-body.
        :type body: dict
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .databases()
            .patch(project=project_id, instance=instance, database=database, body=body)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def delete_database(self, instance: str, database: str, project_id: str) -> None:
        """
        Deletes a database from a Cloud SQL instance.

        :param instance: Database instance ID. This does not include the project ID.
        :type instance: str
        :param database: Name of the database to be deleted in the instance.
        :type database: str
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .databases()
            .delete(project=project_id, instance=instance, database=database)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    @GoogleBaseHook.operation_in_progress_retry()
    def export_instance(self, instance: str, body: Dict, project_id: str) -> None:
        """
        Exports data from a Cloud SQL instance to a Cloud Storage bucket as a SQL dump
        or CSV file.

        :param instance: Database instance ID of the Cloud SQL instance. This does not include the
            project ID.
        :type instance: str
        :param body: The request body, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/export#request-body
        :type body: dict
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        response = (
            self.get_conn()  # noqa # pylint: disable=no-member
            .instances()
            .export(project=project_id, instance=instance, body=body)
            .execute(num_retries=self.num_retries)
        )
        operation_name = response["name"]
        self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)

    @GoogleBaseHook.fallback_to_default_project_id
    def import_instance(self, instance: str, body: Dict, project_id: str) -> None:
        """
        Imports data into a Cloud SQL instance from a SQL dump or CSV file in
        Cloud Storage.

        :param instance: Database instance ID. This does not include the
            project ID.
        :type instance: str
        :param body: The request body, as described in
            https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/export#request-body
        :type body: dict
        :param project_id: Project ID of the project that contains the instance. If set
            to None or missing, the default project_id from the Google Cloud connection is used.
        :type project_id: str
        :return: None
        """
        try:
            response = (
                self.get_conn()  # noqa # pylint: disable=no-member
                .instances()
                .import_(project=project_id, instance=instance, body=body)
                .execute(num_retries=self.num_retries)
            )
            operation_name = response["name"]
            self._wait_for_operation_to_complete(project_id=project_id, operation_name=operation_name)
        except HttpError as ex:
            raise AirflowException(f'Importing instance {instance} failed: {ex.content}')

    def _wait_for_operation_to_complete(self, project_id: str, operation_name: str) -> None:
        """
        Waits for the named operation to complete - checks status of the
        asynchronous call.

        :param project_id: Project ID of the project that contains the instance.
        :type project_id: str
        :param operation_name: Name of the operation.
        :type operation_name: str
        :return: None
        """
        service = self.get_conn()
        while True:
            operation_response = (
                service.operations()  # noqa # pylint: disable=no-member
                .get(project=project_id, operation=operation_name)
                .execute(num_retries=self.num_retries)
            )
            if operation_response.get("status") == CloudSqlOperationStatus.DONE:
                error = operation_response.get("error")
                if error:
                    # Extracting the errors list as string and trimming square braces
                    error_msg = str(error.get("errors"))[1:-1]
                    raise AirflowException(error_msg)
                # No meaningful info to return from the response in case of success
                return
            time.sleep(TIME_TO_SLEEP_IN_SECONDS)


CLOUD_SQL_PROXY_DOWNLOAD_URL = "https://dl.google.com/cloudsql/cloud_sql_proxy.{}.{}"
CLOUD_SQL_PROXY_VERSION_DOWNLOAD_URL = (
    "https://storage.googleapis.com/cloudsql-proxy/{}/cloud_sql_proxy.{}.{}"
)

GCP_CREDENTIALS_KEY_PATH = "extra__google_cloud_platform__key_path"
GCP_CREDENTIALS_KEYFILE_DICT = "extra__google_cloud_platform__keyfile_dict"


class CloudSqlProxyRunner(LoggingMixin):
    """
    Downloads and runs cloud-sql-proxy as subprocess of the Python process.

    The cloud-sql-proxy needs to be downloaded and started before we can connect
    to the Google Cloud SQL instance via database connection. It establishes
    secure tunnel connection to the database. It authorizes using the
    Google Cloud credentials that are passed by the configuration.

    More details about the proxy can be found here:
    https://cloud.google.com/sql/docs/mysql/sql-proxy

    :param path_prefix: Unique path prefix where proxy will be downloaded and
        directories created for unix sockets.
    :type path_prefix: str
    :param instance_specification: Specification of the instance to connect the
        proxy to. It should be specified in the form that is described in
        https://cloud.google.com/sql/docs/mysql/sql-proxy#multiple-instances in
        -instances parameter (typically in the form of ``<project>:<region>:<instance>``
        for UNIX socket connections and in the form of
        ``<project>:<region>:<instance>=tcp:<port>`` for TCP connections.
    :type instance_specification: str
    :param gcp_conn_id: Id of Google Cloud connection to use for
        authentication
    :type gcp_conn_id: str
    :param project_id: Optional id of the Google Cloud project to connect to - it overwrites
        default project id taken from the Google Cloud connection.
    :type project_id: str
    :param sql_proxy_version: Specific version of SQL proxy to download
        (for example 'v1.13'). By default latest version is downloaded.
    :type sql_proxy_version: str
    :param sql_proxy_binary_path: If specified, then proxy will be
        used from the path specified rather than dynamically generated. This means
        that if the binary is not present in that path it will also be downloaded.
    :type sql_proxy_binary_path: str
    """

    def __init__(
        self,
        path_prefix: str,
        instance_specification: str,
        gcp_conn_id: str = 'google_cloud_default',
        project_id: Optional[str] = None,
        sql_proxy_version: Optional[str] = None,
        sql_proxy_binary_path: Optional[str] = None,
    ) -> None:
        super().__init__()
        self.path_prefix = path_prefix
        if not self.path_prefix:
            raise AirflowException("The path_prefix must not be empty!")
        self.sql_proxy_was_downloaded = False
        self.sql_proxy_version = sql_proxy_version
        self.download_sql_proxy_dir = None
        self.sql_proxy_process = None  # type: Optional[Popen]
        self.instance_specification = instance_specification
        self.project_id = project_id
        self.gcp_conn_id = gcp_conn_id
        self.command_line_parameters = []  # type:  List[str]
        self.cloud_sql_proxy_socket_directory = self.path_prefix
        self.sql_proxy_path = (
            sql_proxy_binary_path if sql_proxy_binary_path else self.path_prefix + "_cloud_sql_proxy"
        )
        self.credentials_path = self.path_prefix + "_credentials.json"
        self._build_command_line_parameters()

    def _build_command_line_parameters(self) -> None:
        self.command_line_parameters.extend(['-dir', self.cloud_sql_proxy_socket_directory])
        self.command_line_parameters.extend(['-instances', self.instance_specification])

    @staticmethod
    def _is_os_64bit() -> bool:
        return platform.machine().endswith('64')

    def _download_sql_proxy_if_needed(self) -> None:
        if os.path.isfile(self.sql_proxy_path):
            self.log.info("cloud-sql-proxy is already present")
            return
        system = platform.system().lower()
        processor = "amd64" if CloudSqlProxyRunner._is_os_64bit() else "386"
        if not self.sql_proxy_version:
            download_url = CLOUD_SQL_PROXY_DOWNLOAD_URL.format(system, processor)
        else:
            download_url = CLOUD_SQL_PROXY_VERSION_DOWNLOAD_URL.format(
                self.sql_proxy_version, system, processor
            )
        proxy_path_tmp = self.sql_proxy_path + ".tmp"
        self.log.info("Downloading cloud_sql_proxy from %s to %s", download_url, proxy_path_tmp)
        response = requests.get(download_url, allow_redirects=True)
        # Downloading to .tmp file first to avoid case where partially downloaded
        # binary is used by parallel operator which uses the same fixed binary path
        with open(proxy_path_tmp, 'wb') as file:
            file.write(response.content)
        if response.status_code != 200:
            raise AirflowException(
                "The cloud-sql-proxy could not be downloaded. Status code = {}. "
                "Reason = {}".format(response.status_code, response.reason)
            )

        self.log.info("Moving sql_proxy binary from %s to %s", proxy_path_tmp, self.sql_proxy_path)
        shutil.move(proxy_path_tmp, self.sql_proxy_path)
        os.chmod(self.sql_proxy_path, 0o744)  # Set executable bit
        self.sql_proxy_was_downloaded = True

    @provide_session
    def _get_credential_parameters(self, session: Session) -> List[str]:
        connection = session.query(Connection).filter(Connection.conn_id == self.gcp_conn_id).first()
        session.expunge_all()
        if connection.extra_dejson.get(GCP_CREDENTIALS_KEY_PATH):
            credential_params = ['-credential_file', connection.extra_dejson[GCP_CREDENTIALS_KEY_PATH]]
        elif connection.extra_dejson.get(GCP_CREDENTIALS_KEYFILE_DICT):
            credential_file_content = json.loads(connection.extra_dejson[GCP_CREDENTIALS_KEYFILE_DICT])
            self.log.info("Saving credentials to %s", self.credentials_path)
            with open(self.credentials_path, "w") as file:
                json.dump(credential_file_content, file)
            credential_params = ['-credential_file', self.credentials_path]
        else:
            self.log.info(
                "The credentials are not supplied by neither key_path nor "
                "keyfile_dict of the gcp connection %s. Falling back to "
                "default activated account",
                self.gcp_conn_id,
            )
            credential_params = []

        if not self.instance_specification:
            project_id = connection.extra_dejson.get('extra__google_cloud_platform__project')
            if self.project_id:
                project_id = self.project_id
            if not project_id:
                raise AirflowException(
                    "For forwarding all instances, the project id "
                    "for Google Cloud should be provided either "
                    "by project_id extra in the Google Cloud connection or by "
                    "project_id provided in the operator."
                )
            credential_params.extend(['-projects', project_id])
        return credential_params

    def start_proxy(self) -> None:
        """
        Starts Cloud SQL Proxy.

        You have to remember to stop the proxy if you started it!
        """
        self._download_sql_proxy_if_needed()
        if self.sql_proxy_process:
            raise AirflowException(f"The sql proxy is already running: {self.sql_proxy_process}")
        else:
            command_to_run = [self.sql_proxy_path]
            command_to_run.extend(self.command_line_parameters)
            self.log.info("Creating directory %s", self.cloud_sql_proxy_socket_directory)
            Path(self.cloud_sql_proxy_socket_directory).mkdir(parents=True, exist_ok=True)
            command_to_run.extend(self._get_credential_parameters())  # pylint: disable=no-value-for-parameter
            self.log.info("Running the command: `%s`", " ".join(command_to_run))
            self.sql_proxy_process = Popen(command_to_run, stdin=PIPE, stdout=PIPE, stderr=PIPE)
            self.log.info("The pid of cloud_sql_proxy: %s", self.sql_proxy_process.pid)
            while True:
                line = (
                    self.sql_proxy_process.stderr.readline().decode('utf-8')
                    if self.sql_proxy_process.stderr
                    else ""
                )
                return_code = self.sql_proxy_process.poll()
                if line == '' and return_code is not None:
                    self.sql_proxy_process = None
                    raise AirflowException(
                        f"The cloud_sql_proxy finished early with return code {return_code}!"
                    )
                if line != '':
                    self.log.info(line)
                if "googleapi: Error" in line or "invalid instance name:" in line:
                    self.stop_proxy()
                    raise AirflowException(f"Error when starting the cloud_sql_proxy {line}!")
                if "Ready for new connections" in line:
                    return

    def stop_proxy(self) -> None:
        """
        Stops running proxy.

        You should stop the proxy after you stop using it.
        """
        if not self.sql_proxy_process:
            raise AirflowException("The sql proxy is not started yet")
        else:
            self.log.info("Stopping the cloud_sql_proxy pid: %s", self.sql_proxy_process.pid)