【Mysql】Python连接Mysql进行增删改查

• 一、Python连接Mysql
• • 1.1、python导入pymysql模块：
  • 1.2、连接mysql操作
  • • 1.2.1、首先导入模块：
    • 1.2.2、连接数据库：
    • 1.2.3、新增insert动作：
    • 1.2.4、更新update操作：
    • 1.2.5、查询select语句：
    • 1.2.6、执行sql结果

日常编码测试过程中经常会进行连接数据库进行增、删、改、查操作，接下来就会介绍如何进行封装操作

pip install PyMySQL

我这里安装的是1.0.2版本

import pymysql as my
import time
from Demaxiya.openpay.untils.untils_api import read_data_yaml

class Mysql(object):

    def __init__(self, db="risk"):
        # 数据库连接重试功能和连接超时功能的DB连接
        self._max_retries_count = 2  # 设置最大重试次数
        self._conn_retries_count = 0  # 初始重试次数
        self._conn_timeout = 100  # 连接超时时间为10秒
        # mysql数据库连接
        if db == 'risk':
            self.db_host = read_data_yaml()['test']['mysql']['risk']['host']
            self.username = read_data_yaml()['test']['mysql']['risk']['user']
            self.pw = read_data_yaml()['test']['mysql']['risk']['password']
            self.dbname = read_data_yaml()['test']['mysql']['risk']['name']
        if db == 'afi':
            self.db_host = read_data_yaml()['test']['mysql']['afi']['host']
            self.username = read_data_yaml()['test']['mysql']['afi']['user']
            self.pw = read_data_yaml()['test']['mysql']['afi']['password']
            self.dbname = read_data_yaml()['test']['mysql']['afi']['name']
        if db == 'aku':
            self.db_host = read_data_yaml()['test']['mysql']['aku']['host']
            self.username = read_data_yaml()['test']['mysql']['aku']['user']
            self.pw = read_data_yaml()['test']['mysql']['aku']['password']
            self.dbname = read_data_yaml()['test']['mysql']['aku']['name']
        while self._conn_retries_count <= self._max_retries_count:
            try:
                self.db = my.connect(
                    host=self.db_host,
                    user=self.username,
                    password=self.pw,
                    db=self.dbname,
                    charset='utf8',
                    connect_timeout=self._conn_timeout)
                # cursor = self.db.cursor()#创建游标
                self.cursor = self.db.cursor(cursor=my.cursors.DictCursor)  # 查询结果以字典形式返回
                print("mysql连接success")
                break
            except:
                print("mysql连接false")
                self._conn_retries_count += 1
            time.sleep(2)
            continue

 def inserDB(self, sql):
        # 插入数据库操作
        try:
            # 执行sql
            self.cursor.execute(sql)
            self.db.commit()
            print("数据库插入操作-成功")
        except Exception as e:
            # 异常捕获
            raise e
            # 发生错误时回滚
            self.db.rollback()

    def updateDB(self, sql):
        # ''' 更新数据库操作 '''
        try:
            # 执行sql
            self.cursor.execute(sql)
            self.db.commit()
            print("更新数据库操作-成功")
        except Exception as e:
            # 异常捕获
            raise e
            # 发生错误时回滚
            self.db.rollback()

    def selectDB(self, sql):
        # ''' 数据库查询 '''
        while self._conn_retries_count <= self._max_retries_count:
            try:
                self.cursor.execute(sql)  # 返回 查询数据 条数 可以根据 返回值 判定处理结果
                data = self.cursor.fetchall()  # 返回查询记录列表
                if data == ():
                    return "结果为空了啊，请查看sql正确"
                    break
                else:
                    return [item[key] for item in data for key in item]
                    # print([item[key] for item in data for key in item])
                    break
            except Exception as e:
                raise e
                continue
        self.cursor.close()
        self.db.close()

if __name__ == "__main__":
    test = 1
    sql = "select uid from `test`.`t_1` where test='{}';".format(test)
    print(sql)
    test = Mysql('test').selectDB(sql=sql)
    此时是以列表结果形式返回的：所以要test[0]这样就可以获得查询结果了