阅读量:0
import subprocess class MySQLRemoteConnection: def __init__(self, host, user, password, database): """ 初始化远程MySQL连接类。 :param host: 数据库主机地址 :param user: 数据库用户名 :param password: 数据库密码 :param database: 要连接的数据库名称 """ self.host = host self.user = user self.password = password self.database = database def connect(self): """ 连接到MySQL数据库。 :return: None """ # 构建连接命令 command = f"mysql h {self.host} u {self.user} p'{self.password}' {self.database}" # 执行命令 try: # 使用subprocess.Popen来执行命令，并捕获输出 process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() # 检查是否有错误输出 if stderr: raise Exception(f"Error connecting to MySQL: {stderr.decode().strip()}") print("Connected to MySQL database successfully.") except Exception as e: print(f"An error occurred: {e}") def execute_query(self, query): """ 执行一个查询。 :param query: 要执行的SQL查询 :return: None """ # 构建查询命令 command = f"mysql h {self.host} u {self.user} p'{self.password}' {self.database} e '{query}'" # 执行命令 try: # 使用subprocess.Popen来执行命令，并捕获输出 process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() # 检查是否有错误输出 if stderr: raise Exception(f"Error executing query: {stderr.decode().strip()}") print("Query executed successfully.") print(stdout.decode().strip()) except Exception as e: print(f"An error occurred: {e}") 使用示例 if __name__ == "__main__": # 创建一个连接实例 mysql_conn = MySQLRemoteConnection('your_host', 'your_user', 'your_password', 'your_database') # 连接到数据库 mysql_conn.connect() # 执行一个查询 mysql_conn.execute_query("SELECT * FROM your_table")在这个类中，connect 方法用于建立与MySQL数据库的连接，而execute_query 方法用于执行SQL查询，此代码假设MySQL客户端已经安装在执行脚本的环境中，并且MySQL服务是可访问的。
在实际使用中，你需要替换'your_host','your_user','your_password', 和'your_database' 为你的数据库的实际连接信息，出于安全考虑，直接在命令行中使用密码可能不是最佳实践，考虑使用更安全的方法来管理数据库凭据。
