airflow/contrib/hooks/gcp_sql_hook.py
def start_proxy(self):
        """
        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("The sql proxy is already running: {}".format(
                self.sql_proxy_process))
        else:
            command_to_run = [self.sql_proxy_path]
            command_to_run.extend(self.command_line_parameters)
            try:
                self.log.info("Creating directory %s",
                              self.cloud_sql_proxy_socket_directory)
                os.makedirs(self.cloud_sql_proxy_socket_directory)
            except OSError:
                # Needed for python 2 compatibility (exists_ok missing)
                pass
            command_to_run.extend(self._get_credential_parameters())
            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')
                return_code = self.sql_proxy_process.poll()
                if line == '' and return_code is not None:
                    self.sql_proxy_process = None
                    raise AirflowException(
                        "The cloud_sql_proxy finished early with return code {}!".format(
                            return_code))
                if line != '':
                    self.log.info(line)
                if "googleapi: Error" in line or "invalid instance name:" in line:
                    self.stop_proxy()
                    raise AirflowException(
                        "Error when starting the cloud_sql_proxy {}!".format(
                            line))
                if "Ready for new connections" in line:
                    return
def start_proxy(self):
        self._download_sql_proxy_if_needed()
        if self.sql_proxy_process:
            raise AirflowException("The sql proxy is already running: {}".format(
                self.sql_proxy_process))
        else:
            command_to_run = [self.sql_proxy_path]
            command_to_run.extend(self.command_line_parameters)
            try:
                self.log.info("Creating directory %s",
                              self.cloud_sql_proxy_socket_directory)
                os.makedirs(self.cloud_sql_proxy_socket_directory)
            except OSError:
                # Needed for python 2 compatibility (exists_ok missing)
                pass
            command_to_run.extend(self._get_credential_parameters())
            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')
                return_code = self.sql_proxy_process.poll()
                if line == '' and return_code is not None:
                    self.sql_proxy_process = None
                    raise AirflowException(
                        "The cloud_sql_proxy finished early with return code {}!".format(
                            return_code))
                if line != '':
                    self.log.info(line)
                if "googleapi: Error" in line or "invalid instance name:" in line:
                    self.stop_proxy()
                    raise AirflowException(
                        "Error when starting the cloud_sql_proxy {}!".format(
                            line))
                if "Ready for new connections" in line:
                    return
def start_proxy ( self ) : self . _download_sql_proxy_if_needed ( ) if self . sql_proxy_process : raise AirflowException ( "The sql proxy is already running: {}" . format ( self . sql_proxy_process ) ) else : command_to_run = [ self . sql_proxy_path ] command_to_run . extend ( self . command_line_parameters ) try : self . log . info ( "Creating directory %s" , self . cloud_sql_proxy_socket_directory ) os . makedirs ( self . cloud_sql_proxy_socket_directory ) except OSError : # Needed for python 2 compatibility (exists_ok missing) pass command_to_run . extend ( self . _get_credential_parameters ( ) ) 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' ) return_code = self . sql_proxy_process . poll ( ) if line == '' and return_code is not None : self . sql_proxy_process = None raise AirflowException ( "The cloud_sql_proxy finished early with return code {}!" . format ( return_code ) ) if line != '' : self . log . info ( line ) if "googleapi: Error" in line or "invalid instance name:" in line : self . stop_proxy ( ) raise AirflowException ( "Error when starting the cloud_sql_proxy {}!" . format ( line ) ) if "Ready for new connections" in line : return
Starts Cloud SQL Proxy.

        You have to remember to stop the proxy if you started it!
Starts Cloud SQL Proxy .