首页 > 数据库 >python ssh连接mysql

python ssh连接mysql

时间:2024-01-18 20:37:40浏览次数:33  
标签:__ python self mysql cursor ssh sql

from sshtunnel import SSHTunnelForwarder
import pymysql
class MySqlSSH:
    def __init__(self):
        self.server = SSHTunnelForwarder(
            ssh_address_or_host=('13.229.92.6', 22),    # ssh host
            ssh_username='lenox',      # ssh 账号
            ssh_password='......',   # ssh 密码
            remote_bind_address=('......', 3306)   # 数据库配置
        )
        # 启动隧道服务
        self.server.start()
        mysql_config = {
            'user': '......',
            'passwd': '......',
            'host': self.server.local_bind_host,
            'port': self.server.local_bind_port,
            'db': 'ph_data_bossjob'
        }
        # 连接数据库
        self.mysql = pymysql.connect(**mysql_config)
        self.cursor = self.mysql.cursor()
    def fetch_one(self, sql):
        # 执行SQL
        self.cursor.execute(sql)
        # 查看结果
        result = self.cursor.fetchone()
        return result
    def fetch_all(self, sql):
        # 执行SQL
        self.cursor.execute(sql)
        # 查看结果
        result = self.cursor.fetchall()
        return result
    def execute_sql(self, sql):
        # 执行SQL
        self.cursor.execute(sql)
        # 提交
        self.mysql.commit()
        # 返回影响行数
        result = "受影响的行数:{}".format(self.cursor.rowcount)
        return result
    def close(self):
        self.cursor.close()  # 关闭查询
        self.mysql.close()
        self.server.close()  # 关闭服务
if __name__ == '__main__':
    cc = None
    try:
        cc = MySqlSSH()
        qq = cc.fetch_all("select count(*) from users where date(created_at)='2024-01-18'")
        print(qq)
    except Exception as e:
        print(e)
        print("fail")
    finally:
        cc.close()

 

标签:__,python,self,mysql,cursor,ssh,sql
From: https://www.cnblogs.com/kylin5201314/p/17973339

相关文章

  • mysql 被删除如何恢复
    恢复ibd使用-------OnDebian/Ubuntu/LinuxMint-------$sudoapt-getinstalltestdisk-------OnCentOS/RHEL/Fedora-------$sudoyuminstalltestdisk-------OnFedora22+-------$sudodnfinstalltestdisk-------OnArchLinux-------$pacman-Ste......
  • CPLEX通过Python API获取Gap值的方法
    写在前面最近在使用Cplex求解模型,尽管Cplex的PythonAPI会自动输出引擎日志,但在多次求解中一次次看引擎日志找Gap值并做实验记录很麻烦,所以需要找到获取Gap值的方法。然而我在Cplex的官方文档中并没有找到这个方法,然后我就一个个去试这些方法,可算是给我试出来了。解决方法在Cpl......
  • aws 服务器 ssh 通过密码连接配置
    背景通过信用卡薅了一年aws服务器的羊毛,1cpu30g固态硬盘,默认服务器不允许root用户使用密码进行ssh连接,因此需要以下的步骤通过密码远程ssh连接到服务器。步骤使用AWS控制台进行网页登录创建root密码sudopasswdroot切换到root用户suroot修改sshd_co......
  • Python - Playwright安装
    前言:Playwright是专门为满足端到端测试的需要而创建的。Playwright支持所有现代渲染引擎,包括Chromium、WebKit(Safari的浏览器引擎)和Firefox。在Windows、Linux和macOS上进行本地测试或在CI上进行测试.与Selenium+driver不同的是,Pw需要使用定制版的浏览器。如果本地......
  • python中的+=
    注意点:就地修改: 使用+=会就地修改可变对象,如列表或字典,而不是创建一个新的对象。这意味着原始对象会被改变。与+运算符的区别:+=与+运算符不同。对于不可变对象(如字符串、元组等),+=通常会创建一个新对象,而不是就地修改。可变对象vs不可变对象:对于可变对象(如列表、字典),+......
  • mysql内置函数
    Mysql常用内置函数数字函数函数名称作用ABS求绝对值SQRT求二次方根MOD求余数CEIL和CEILING两个函数功能相同,都是返回不小于参数的最小整数,即向上取整FLOOR向下取整,返回值转化为一个BIGINTRAND生成一个0~1之间的随机数,传入整数参数是,用来产生重......
  • 波达方向估计(DOA)-Python代码实现MVDR
    https://mp.weixin.qq.com/s/61I1aBTwJ3ykw0uuceLKkQ模拟一个由三根全向天线组成的阵列,然后使用数组来模拟到达阵列的信号。相邻天线之间:1/2波长(也称为“半波长间隔”)。将模拟发射机的信号以一定角度theta到达该阵列。另外在这个接收到的信号中添加噪声。importnumpyasnp......
  • MySQL并行复制死锁源码解析
    最近一个MySQL5.7.21备库告警当天的备份失败,登录上去看的时候发现前一天的备份任务还没有结束,通过查看日志发现无法备份成功的原因是一直无法获取FTWRL锁,登录MySQL查看会话状态发现其中几个复制worker一致处于异常状态,下发STOPSLAVE命令时命令也一直被卡住,当时的会话状态如下:......
  • ES和mysql数据同步
    elasticsearch中的数据来自于mysql数据库,因此mysql数据发生改变时,elasticsearch也必须跟着改变,这个就是elasticsearch与mysql之间的数据同步。常见的数据同步方案有三种:同步调用异步通知监听binlog 同步调用 基本步骤如下:hotel-demo对外提供接口,用来修改ela......
  • 波达方向估计(DOA)-Python代码实现
    https://mp.weixin.qq.com/s/fMGc8ziglySGKr1fY8Jvkw模拟一个由三根全向天线组成的阵列,然后使用数组来模拟到达阵列的信号。相邻天线之间距离为1/2波长(也称为“半波长间隔”)。将模拟发射机的信号以一定角度theta到达该阵列。另外在这个接收到的信号中添加噪声。importnumpy......