使用python连接mysql数据库数据，有以下两种读取数据的方式推荐。
一种是通过游标，及fetch系列方法进行操作，另一种是通过pandas的read_sql()进行读取并操作。各种方法各有优劣，可根据具体情形，择优选择使用。

fetchone/fetchmany/fetchall

获取一条、多条、全部条。

import pymysql

# 数据库相关信息
dbHost = 'xxxxxxx'
dbUser = 'xxx'
dbPassword = '******'
dbName = 'xxx'
dbCharset = 'utf8'
conn = pymysql.connect(host=dbHost, port=3306, user=dbUser, password=dbPassword, db=dbName, charset=dbCharset)


# 获取游标对象
cs = conn.cursor()

# 通过游标对象，执行sql语句,返回值为受影响记录的行数
r = cs.execute('select * from goods')

# 获取一条数据
print(cs.fetchone())  # 第一条数据
print("==============================================")
# 再次执行会获取第二条数据
print(cs.fetchone())  # 第二条数据

# 获取多条数据
print(cs.fetchmany(3))  # 指定条数

# 获取全部数据
print(cs.fetchall())
# 再次执行，获取到的将是一个空元组，因为上边的fetchall已经取完了（游标可以理解为对获取位置的标记）
print(cs.fetchall())  # 当获取完毕，再查询数据返回为（）

# 获取结束后，要有始有终，关闭游标和数据库连接
# 关闭游标
cs.close()
# 关闭连接
conn.close()

pandas.read_sql()

使用pandas库的read_sql()函数获取数据，将得到一个DataFrame。

import pymysql
import pandas as pd

# 数据库相关信息
dbHost = 'xxxxxxx'
dbUser = 'xxx'
dbPassword = '******'
dbName = 'xxx'
dbCharset = 'utf8'
conn = pymysql.connect(host=dbHost, port=3306, user=dbUser, password=dbPassword, db=dbName, charset=dbCharset)

sql = "select xxxxxxxxxxxxxxxxxxxxxxxxxxxx"
df = pd.read_sql(sql, conn)
print(df)

# 关闭连接
conn.close()

🐒，本文中，使用到的工具有：Pycharm，Anaconda，MySQL 5.5，spyder(Anaconda)

什么是 PyMySQL？

PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库，Python2 中则使用 mysqldb。

一、🏔环境准备

1、安装pymysql：

进行Python连接mysql数据库之前，需要先安装一下pymysql。

直接在终端执行下面的命令即可。(在此处我将指定1.0.2版本)

pip install pymysql==1.0.2

2、查询安装：

下载完成后，在终端输入 pip list 即可看到下图：

pip list

可以看到我们的PyMySQL是1.0.2版本的。

3、Anaconda下载pymysql：

打开Anaconda，选择 Environments 点击右上方的的搜索框 输入 pymysql

 点击方框，即可下载

方式一🏔：

        此处我们可以选择 spyder 或者 pycharm 首先为大家介绍一下 spyder 我们只需要直接导入 pymysql 库即可 

方式二🏔： 

        4、Pycharm下载pymysql

打开 Pycharm 选择文件，点击设置，

 下划，选择python解释器，这里我的Pycharm已经配置了Anaconda环境

如果没有查询到 pymysql 可以在 Pycharm 终端中下载 pymysql 库

以上我们的环境就准备好了，下面我们进行编写程序 ☀

 二、🏔代码编写，连接数据库

1、导入数据库表 

import pymysql

 数据库连接：

连接数据库前，请先确认以下事项：

• 连接数据库使用的用户名为 "root" ，密码为 "dai324542",创建了数据库 runoob
• 你可以可以自己设定或者直接使用root用户名及其密码

db = pymysql.connect(host='localhost',
                     user='root',
                     password='dai324542',
                     database='runoob',
                     charset='utf8')
 
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
 
# 使用 execute()  方法执行 SQL 查询 
cursor.execute("SELECT VERSION()")
 
# 使用 fetchone() 方法获取单条数据.
data = cursor.fetchone()
 
print ("数据库连接成功！")
 
# 关闭数据库连接
db.close()

 2、创建数据库表

# 创建表
sql="""CREATE TABLE test (
          FIRST_  CHAR(20) NOT NULL,
          SECOND_  CHAR(20),
          THIRD_ INT,
          FOURTH_ CHAR(1),
          FIFTH_ FLOAT )"""
# 运行sql语句
cursor.execute(sql)

这里我们所运用的sql语句是不是很熟悉了😊

下面即是运行结果了，再mysql中可以刷新看到，我输出了一个提示 victory 

 3、数据库插入操作

此处我只是随便进行了一个举例，通过更改创建表时的操作可以插入不同类型的数据

try:
    sql = "insert into test(FIRST_,SECOND_,THIRD_,FOURTH_,FIFTH_) values ('MAC','MOTH','20','M','2000')"

    # 运行sql语句
    cursor.execute(sql)
    # 修改
    db.commit()
    # 关闭游标
    cursor.close()
    # 关闭连接
    db.close()
    print("victory!")
except:
    print("false")

 4、查询其中一个表的数据

# 查询语句
try:
    cursor = db.cursor()
    sql = "select * from student"
    cursor.execute(sql)
    result = cursor.fetchall()
    for data in result:
        print(data)
except Exception:
    print("查询失败")

 5、删除表中的一条数据

# SQL 删除语句
sql = "DELETE FROM student WHERE Sno='20111107'"
try:
   # 执行SQL语句
   cursor.execute(sql)
   # 向数据库提交
   db.commit()
except:
   # 发生错误时回滚
   db.rollback()
# 关闭连接
db.close()
# 成功提示
print("victory!")

注意：Python中的MySQL默认事务打开，需要我们手动提交事务，否则操作无效

写到这里，这篇博客就又又又结束了，很感谢大家的观看，如果对大家有所帮助希望可以留下一个小小的👍，🙇‍。因才学疏浅，如果各位大佬发现其中存在错误，敬请指出，(ง •_•)ง！

前言：

本文介绍两种写python连接mysql代码的方案：

1、线性代码

2、在框架中做配置模块+进行数据驱动

在操作之前需要先在数据库中，建一个mysql的表

一、python连接mysql线性代码

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/10 10:51
# @Function:
'''
    数据容器：mysql
    操作数据库的步骤
    1、连接数据库，通过connnect函数链接，生成connection对象
    2、定义我们的游标Cursor,再通过我们游标执行脚本并获取结果
    3、关闭连接
'''
import datetime
import time

import pymysql

# 1、建立mysql连接
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', password='root', database='test_cases', charset='utf8')
'''
    常用方法：
    1、cursor()使用当前连接创建并返回游标
    2、commit()提交当前事务
    3、rollback()回滚当前事务
    4、close()关闭当前连接
'''
# 2、建立游标
cur = conn.cursor()
'''
    游标操作方法：
    1、execute()执行数据库查询或命令，将结果从数据库返回给客户端
    2、fetchone()获取结果集的下一行
    3、fetchall()获取结果集的所有行
    4、fetchmany()获取结果集的几行
'''
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# now=datetime.datetime.now(.strftime("%Y-%m-%d %H:%M:%S"))
# 3、执行脚本
cur.execute('select weaid,success from weather')
print(cur.fetchall())
cur.execute('insert into weather(weaid,success,cre_time) values ("2","2","%s")' % now)
cur.execute('commit')
conn.close()

二、python连接mysql工具类+进行数据驱动+配置模块

即：将python连接mysql相关的代码封装起来，形成一个工具类，在需要使用的时候调用

准备工作：1、key_demo——2、cases|tool——3、cases/case.py|tool/MyDB.py

1、MyDB.py工具类

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/18 15:20
# @Function:
import pymysql


class my_db:
    '''
    动作类：获取数据连接，连接ip,端口，账号密码。。
    '''

    # 构造函数
    def __init__(self):
        try:
            self.dbconn = pymysql.connect(host='127.0.0.1', port=3306, user='root', password='root',
                                          database='test_cases', charset='utf8')
        except Exception as e:
            print('初始化数据库连接失败：%s' % e)

    def close(self):
        self.dbconn.close()

    # query=查询语句
    def select_record(self, query):
        # 查询数据
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            result = db_cursor.fetchall()
            return result
        except Exception as e:
            print('数据库查询数据失败：%s' % e)
            db_cursor.close()
            exit()

    # 插入
    def execute_insert(self, query):
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            db_cursor.execute('commit')
            return True
        except Exception as e:
            print('数据库插入数据失败：%s' % e)
            # 事务回滚
            db_cursor.execute('rollback')
            db_cursor.close()
            exit()

2、case.py数据驱动

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 针对数据库中的数据做数据驱动
import time
import unittest
from ddt import ddt, data, unpack

from key_demo.tool.MyDB import my_db

# 必须要实例化，不然会报TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(*testdb.select_record('select weaid,success from weather'))
    @unpack
    def test_1(self, weaid, success):
        print(weaid, success)
        testdb.execute_insert('insert into weather(weaid,success,cre_time) values ("2","2","%s")' % now)


if __name__ == '__main__':
    unittest.main()

运行结果：

这样写的插入语句会被调用多次，要注意@data中执行的操作具体执行次数

换了一个写法，先插入，再查询

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 针对数据库中的数据做数据驱动
import time
import unittest
from ddt import ddt, data, unpack

from key_demo.tool.MyDB import my_db

# 必须要实例化，不然会报TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(testdb.execute_insert('insert into weather(weaid,success,cre_time)values("2","2","%s")' % now))
    # @unpack
    def test_1(self,t):
        print(t)
        res = testdb.select_record('select weaid,success from weather')
        print(res)


if __name__ == '__main__':
    unittest.main()

代码编写的时候有几个注意事项：

1、只执行一条语句，返回结果也只有一个，不需要@unpack解析

2、MyDB.py中定义有返回值，还有ddt中定义的语法return func(self, *args, **kwargs),不能缺少参数

运行结果： 

还有一种方式：

@ddt
class testcase(unittest.TestCase):
    @data(['2', '2'], ['1', '1'])
    @unpack
    def test_1(self, value1, value2):
        testdb.execute_insert('insert into weather(weaid,success,cre_time)values(' + value1 + ',' + value2 + ',%s)' % now)
        res = testdb.select_record('select weaid,success from weather')
        print(res)

可优化的点：data里面的值可以进行数据驱动测试，将数据和代码分离

数据放在yaml或者csv中读取

3、通过配置模块优化数据库连接

3、1定义配置项dbconfig.conf

1.key_demo——2.key_demo/config——3、key_demo/config/dbconfig.conf

#测试环境
[TESTDB]
host=127.0.0.1
port=3306
user=root
password=root
db=test_cases
charset=utf8

[DEVDB]
host=192.168.1.2
port=3306
user=root
password=root
db=test_cases1
charset=utf8

#生产环境
[prdDB]
host=192.168.1.4
port=3306
user=root
password=root
db=test_cases2
charset=utf8

3、2改造工具类的构造函数

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/18 15:20
# @Function:
import pymysql

import configparser


class my_db:
    '''
    动作类：获取数据连接，连接ip,端口，账号密码。。
    '''

    # 构造函数
    def __init__(self, config_file, db):
        # 实例化configparser
        config = configparser.ConfigParser()
        # 从配置文件中读取数据库相关信息
        config.read(config_file)
        host = config[db]['host']
        port = int(config[db]['port'])
        user = config[db]['user']
        password = config[db]['password']
        database = config[db]['database']
        charset = config[db]['charset']
        try:
            self.dbconn = pymysql.connect(host, port, user, password, database, charset)
        except Exception as e:
            print('初始化数据库连接失败：%s' % e)

    def close(self):
        self.dbconn.close()

    # query=查询语句
    def select_record(self, query):
        # 查询数据
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            result = db_cursor.fetchall()
            return result
        except Exception as e:
            print('数据库查询数据失败：%s' % e)
            db_cursor.close()
            exit()

    # 插入
    def execute_insert(self, query):
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            db_cursor.execute('commit')
            return True
        except Exception as e:
            print('数据库插入数据失败：%s' % e)
            # 事务回滚
            db_cursor.execute('rollback')
            db_cursor.close()
            exit()

注意事项：configparser配置对象的使用

3、3在用例中的使用

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 针对数据库中的数据做数据驱动
import time
import unittest
from ddt import ddt, data, unpack

from key_demo.tool.MyDB import my_db

# 必须要实例化，不然会报TypeError: select_record() missing 1 required positional argument: 'query'
# testdb = my_db('key_demo/config/dbconfig.conf','TESTDB')
config_file='key_demo/config/dbconfig.conf'
db='TESTDB'
testdb=my_db(config_file,db)
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(testdb.execute_insert('insert into weather(weaid,success,cre_time)values("2","2","%s")' % now))
    # @unpack
    def test_1(self, t):
        print(t)
        res = testdb.select_record('select weaid,success from weather')
        print(res)


if __name__ == '__main__':
    unittest.main()

3、4运行结果