Da

✏️ Editing: mysql.py

import json
import posixpath
from contextlib import closing

from parallels.core import messages, MigrationError, safe_str_decode
from parallels.core import safe_format
from parallels.core.logging import get_logger
from parallels.core.runners.base import BaseRunner
from parallels.core.runners.entities import ExecutionOptions
from parallels.core.thirdparties import pymysql
from parallels.core.utils.common.sql_utils import format_sql_results_for_log
from parallels.core.utils.json_utils import encode_json
from parallels.core.utils.pmm.utils import convert_sql_query_for_perl
from parallels.core.utils.thirdparties import packages
from parallels.core.utils.thirdparties.deployer import ThirdpartyDeployer

logger = get_logger(__name__)

def mysql_query_local(connection_args, query_str, query_args=None, is_result_as_dict=False):
    """Execute given MySQL query from the local server

:type connection_args: dict
    :type query_str: str
    :type query_args: dict | None
    :type is_result_as_dict: bool
    :rtype: list | dict
    """
    log_query = query_str
    if query_args is not None:
        log_query = log_query % {k: "'%s'" % (v,) for k, v in query_args.items()}
    log_server = connection_args['host'] if 'host' in connection_args else 'localhost'
    with logger.debug_level_context(messages.SQL_QUERY, server=log_server):
        logger.fdebug(messages.SQL_QUERY_QUERY, query=log_query)
        with closing(pymysql.connect(**connection_args)) as connection:
            with connection.cursor(pymysql.cursors.DictCursor if is_result_as_dict else None) as cursor:
                cursor.execute(query_str, query_args)
                results = cursor.fetchall()
        logger.fdebug(messages.SQL_QUERY_RESULTS, results=format_sql_results_for_log(results))
    return results

def escape_args_list(args, prefix):
    """Prepare list of placeholders for SQL query and corresponding list of values for SQL query formatting

:type args: list[str | int]
    :type prefix: str
    :rtype: list[str], dict[str]
    """
    placeholders = []
    values = {}
    for i, arg in enumerate(args):
        key = '%s_%s' % (prefix, i)
        placeholders.append('%(' + key + ')s')
        values[key] = arg
    return placeholders, values

def mysql_query_with_agent_unix(server, connection_args, query_str, query_args=None):
    """Execute given MySQL query from remote Unix server

:type server: parallels.core.connections.server.Server
    :type connection_args: dict
    :type query_str: str
    :type query_args: dict | None
    :rtype: dict
    """
    mysql_agent = ThirdpartyDeployer.get_instance().deploy(packages.MySQLAgent, server)

log_query = query_str
    if query_args is not None:
        log_query = log_query % {k: "'%s'" % (v,) for k, v in query_args.items()}
    log_server = connection_args['host'] if 'host' in connection_args else 'localhost'
    with logger.debug_level_context(
        messages.SQL_QUERY_REMOTE,
        physical_server=server.description(), mysql_server=log_server
    ):
        logger.fdebug(messages.SQL_QUERY_QUERY, query=log_query)
        connection_settings = dict(
            user=connection_args.get('user'),
            password=connection_args.get('password'),
            unix_socket=connection_args.get('unix_socket'),
            database=connection_args.get('database'),
            port=connection_args.get('port')
        )

query_template, query_template_params = convert_sql_query_for_perl(
            query_str, query_args, quote_questions=False
        )

stdin_content = encode_json(dict(
            query=query_template,
            query_args=query_template_params,
            connection_settings=connection_settings
        ))

with server.runner() as runner:
            assert isinstance(runner, BaseRunner)
            stdout = runner.execute_command(
                "cd {agent_dir} && perl {mysql_agent}", dict(
                    agent_dir=posixpath.dirname(mysql_agent),
                    mysql_agent=mysql_agent
                ),
                ExecutionOptions(stdin_content=stdin_content)
            ).stdout

try:
                agent_response = json.loads(safe_str_decode(stdout))
            except Exception as e:
                logger.exception()
                raise MigrationError(safe_format(
                    messages.SQL_QUERY_FAILED_TO_PARSE_JSON,
                    reason=e, json=stdout
                ))

if agent_response.get('status') != 'ok':
                raise MigrationError(safe_format(
                    messages.SQL_QUERY_FAILED,
                    reason=agent_response['message']
                ))
            else:
                results = agent_response['rowset']
                logger.fdebug(messages.SQL_QUERY_RESULTS, results=format_sql_results_for_log(results))
                return results

📁 Sid Gifari File Manager