【pymysql的基本使用】

0. 介绍

本文主要介绍如何使用pymysql库来操作mysql数据库，包含docker安装MySQL和对Mysql的各种操作。

参考链接：

Welcome to PyMySQL’s documentation! — PyMySQL 0.7.2 documentation

Python3 MySQL 数据库连接 – PyMySQL 驱动 | 菜鸟教程

Python之pymysql详解_LinWoW的博客-CSDN博客_pymysql原理

MySQL 教程 | 菜鸟教程

1. 安装Mysql

从docker中拉取MySQL镜像

# 从docker仓库中拉取最新的MySQL镜像
docker pull mysql 
# 从docker仓库中拉取指定版本的Mysql镜像
docker pull mysql:5.7

创建MySQL容器

docker run -d --name=MYSQL_NAME -p 3306:3306 -v mysql-data:/val/lib/mysql -e MYSQL_ROOT_PASSWORD=your_password mysql

其中各个参数的含义如下：

• -d：以分离模式运行此容器，以便后台运行
• --name：容器实例name
• -p：将Mysql容器的3306端口绑定到主机的3306端口上，这样通过主机的端口就可以访问MySQL容器
• -v：将容器卷（/var/lib/mysql)内的文件夹绑定到主机的mysql-data路径下
• -e：设置环境变量
• mysql：创建该容器镜像的名称

连接MySQL容器

• 使用docker ps -a，查看运行MySQL的container ID，通过container ID连接

• 使用容器名称连接

docker exec -it MYSQL_NAME bash

登录MySQL

mysql -u root -p 

输入root用户的password，即可登录MySQL数据库。

新建database

在MySQL中，使用如下命令新建一个database用于后续的实验。

 CREATE DATABASE test_db;

 返回Query OK，即创建test_db database成功，也可以通过如下命令查看所有的database。

SHOW DATABASES;

通过如下命令，进入新建的test_db database中 ，后续的实验都是在这个database中进行。

use test_db;

2. pymysql操作MySQL

2.1 安装pymysql库

pymysql是一个纯python库，可以直接使用pip安装，命令如下：

pip install pymysql

2.2 pymysql的基本操作

通过pymysql对MySQL数据库的常见操作包括：数据库连接、创建database、新建table，向table中插入数据，删除数据，修改数据和查询数据等。

本文将以图像数据存取到MySQL数据库作为例子，描述上述的相关操作如何实现。

连接MySQL数据库

通过上述连接MySQL容器，登录MySQL等操作可以确认MySQL容器是启动的，然后通过如下代码就可以实现MySQL数据库的连接。

MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = 3306
MYSQL_USER = "root"
MYSQL_PWD = "root用户的密码"
MYSQL_DB = "test_db"
#创建与数据库的连接
db=pymysql.connect(host=MYSQL_HOST, user=MYSQL_USER, port=MYSQL_PORT, password=MYSQL_PWD,
                                    database=MYSQL_DB,
                                    local_infile=True)
#创建游标对象cursor
cursor=db.cursor()

新建table

首先，新建的这个table是用来保存图像数据，包括图像id，图像名和图像二进制数据。通过参考MySQL的数据类型，可以使用如下命令来新建用于保存图像的table。

sql = "create table if not exists " + table_name + "(image_id INT PRIMARY KEY auto_increment, image_path TEXT, image_data MEDIUMBLOB not null);"
cursor.execute(sql)

使用到的数据类型介绍如下：

可以在MySQL中使用如下命令，查看和删除table

show tables;  # 查看当前database中所有的tables
drop table table名称; # 用来删除指定的table

 插入单条数据

对于图像的数据内容，需要使用base64库编码，得到二进制形式的文本数据。load_one_data_to_mysql函数将输入的图像数据，经过encode_image_base64函数编码，然后插入到指定的table中。

    def encode_image_base64(self, image_name):
        with open(image_name, "rb") as f:
            img_data = f.read()
            base64_data = base64.b64encode(img_data)
        return base64_data
 
    def load_one_data_to_mysql(self, table_name, file_name):
        img_data = self.encode_image_base64(file_name)
        self.test_connection()
        sql = "insert into " + table_name + " (image_path,image_data) values (%s,%s);"
        try:
            self.cursor.execute(sql, (file_name.encode(), img_data))
            self.conn.commit()
            LOGGER.debug(f"MYSQL loads one data to table: {table_name} successfully")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

同样可以，在终端使用如下命令插入当前table中，有效元组的个数

select count(*) from table名称;

 插入多条数据

在上面的函数中输入是一张图像的名称，当需要一次性插入多张图像数据时，可以输入图像名列表。

    def load_data_to_mysql(self, table_name, image_name_list):
        data = list()
        for img_name in image_name_list:
            img_data = self.encode_image_base64(img_name)
            data.append((img_name, img_data))
        # Batch insert (Milvus_ids, img_path) to mysql
        self.test_connection()
        sql = "insert into " + table_name + " (image_path,image_data) values (%s,%s);"
        try:
            self.cursor.executemany(sql, data)
            self.conn.commit()
            LOGGER.debug(f"MYSQL loads batch data to table: {table_name} successfully")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

 在终端查询之后，得到当前table中包含28条元组

查询数据

MySQL中使用SELECT语句来查询数据，具体语法如下所示：

 在这里，列举了几个该实验常见可用的SQL语句

查询现在table中元组的个数

SELECT COUNT(*) FROM image;

查询所有的image_path内容

SELECT image_path FROM image;

 查询指定image_id的image_path和image_data内容

SELECT image_path, image_data FROM image WHERE image_id=3;

 接下来实现，输入image_id查询image_data数据内容，并将其使用指定文件名保存下来，可以直接查看。

    def decode_image_base64(self, base64_data, filename):
        with open(filename, "wb") as f:
            img_data = base64.b64decode(base64_data)
            f.write(img_data)
 
    def query_by_image_id(self, table_name, image_id, filename):
        # Get the image_path and image_data according to the image_id
        self.test_connection()
        sql = "select image_data from " + table_name + " where image_id=%s;"
        try:
            self.cursor.execute(sql, (image_id))
            results = self.cursor.fetchone()
            LOGGER.debug("MYSQL query by image_id.")
            self.decode_image_base64(results[0], filename)
            LOGGER.debug("Decode image data success.")
            return "ok"
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

修改（更新）数据

MySQL使用UPDATA语句来更新元组的内容，具体语法如下：

 在这里，可以基于给定的image_id和新的图像名来对table中对应的image_id内容进行更新。

    def updata_by_image_id(self, image_id, filename):
 
        self.test_connection()
        sql = "update " + table_name + " set image_path=%s, image_data=%s where image_id=%s;"
        base64_data = self.encode_image_base64(filename)
        try:
            self.cursor.execute(sql, (filename.encode(), base64_data, image_id))
            self.conn.commit()
            LOGGER.debug("MYSQL updata by image_id.")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

 为了验证该update操作成功，可以在其操作前后分别查询image_id对应image_data保存图像是否发生变化。

删除数据

MySQL使用DELETE语句来删除元组的内容，具体语法如下：

 在本次实验中，可以基于给定的image_id来删除table中对应元组的内容。首先执行如下命令查看当前table中元组的个数，执行删除操作之后，再查看元组格式是否减少一个。

    def delete_by_image_id(self, table_name, image_id):
        # Delete all the data in mysql table
        self.test_connection()
        sql = 'delete from ' + table_name + ' where image_id=%s;'
        try:
            self.cursor.execute(sql, (image_id))
            self.conn.commit()
            LOGGER.debug(f"MYSQL delete data by image_id in table:{table_name}")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

 

删除table

最后，当这个table不再需要时候，可以使用如下的语句将table数据表删除。

DROP TABLE table名称;

    def delete_table(self, table_name):
        # Delete mysql table if exists
        self.test_connection()
        sql = "drop table if exists " + table_name + ";"
        try:
            self.cursor.execute(sql)
            LOGGER.debug(f"MYSQL delete table:{table_name}")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

 到这里，整个例子就演示完毕了。

3. 全部代码

class MySQLHelper():
    """
    Say something about the ExampleCalass...
    Args:
        args_0 (`type`):
        ...
    """
    def __init__(self):
        self.conn = pymysql.connect(host=MYSQL_HOST, user=MYSQL_USER, port=MYSQL_PORT, password=MYSQL_PWD,
                                    database=MYSQL_DB,
                                    local_infile=True)
        self.cursor = self.conn.cursor()
 
    def test_connection(self):
        try:
            self.conn.ping()
        except Exception:
            self.conn = pymysql.connect(host=MYSQL_HOST, user=MYSQL_USER, port=MYSQL_PORT, password=MYSQL_PWD,
                                    database=MYSQL_DB,local_infile=True)
            self.cursor = self.conn.cursor()
 
    def create_mysql_table(self, table_name):
        # Create mysql table if not exists
        self.test_connection()
        sql = "create table if not exists " + table_name + "(image_id INT PRIMARY KEY auto_increment, image_path TEXT NOT NULL, image_data MEDIUMBLOB not null);"
        try:
            self.cursor.execute(sql)
            LOGGER.debug(f"MYSQL create table: {table_name} with sql: {sql}")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
    def encode_image_base64(self, image_name):
        with open(image_name, "rb") as f:
            img_data = f.read()
            base64_data = base64.b64encode(img_data)
        return base64_data
 
    def load_one_data_to_mysql(self, table_name, file_name):
        img_data = self.encode_image_base64(file_name)
        self.test_connection()
        sql = "insert into " + table_name + " (image_path,image_data) values (%s,%s);"
        try:
            self.cursor.execute(sql, (file_name.encode(), img_data))
            self.conn.commit()
            LOGGER.debug(f"MYSQL loads one data to table: {table_name} successfully")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
 
    def load_data_to_mysql(self, table_name, image_name_list):
        data = list()
        for img_name in image_name_list:
            img_data = self.encode_image_base64(img_name)
            data.append((img_name, img_data))
        # Batch insert (Milvus_ids, img_path) to mysql
        self.test_connection()
        sql = "insert into " + table_name + " (image_path,image_data) values (%s,%s);"
        try:
            self.cursor.executemany(sql, data)
            self.conn.commit()
            LOGGER.debug(f"MYSQL loads batch data to table: {table_name} successfully")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
    def decode_image_base64(self, base64_data, filename):
        with open(filename, "wb") as f:
            img_data = base64.b64decode(base64_data)
            f.write(img_data)
 
    def query_by_image_id(self, table_name, image_id, filename):
        # Get the image_path and image_data according to the image_id
        self.test_connection()
        sql = "select image_data from " + table_name + " where image_id=%s;"
        try:
            self.cursor.execute(sql, (image_id))
            results = self.cursor.fetchone()
            LOGGER.debug("MYSQL query by image_id.")
            self.decode_image_base64(results[0], filename)
            LOGGER.debug("Decode image data success.")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
    
    def updata_by_image_id(self, image_id, filename):
 
        self.test_connection()
        sql = "update " + table_name + " set image_path=%s, image_data=%s where image_id=%s;"
        base64_data = self.encode_image_base64(filename)
        try:
            self.cursor.execute(sql, (filename.encode(), base64_data, image_id))
            self.conn.commit()
            LOGGER.debug("MYSQL updata by image_id.")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
    def delete_by_image_id(self, table_name, image_id):
        # Delete all the data in mysql table
        self.test_connection()
        sql = 'delete from ' + table_name + ' where image_id=%s;'
        try:
            self.cursor.execute(sql, (image_id))
            self.conn.commit()
            LOGGER.debug(f"MYSQL delete data by image_id in table:{table_name}")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
 
    def delete_table(self, table_name):
        # Delete mysql table if exists
        self.test_connection()
        sql = "drop table if exists " + table_name + ";"
        try:
            self.cursor.execute(sql)
            LOGGER.debug(f"MYSQL delete table:{table_name}")
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)
 
    def count_table(self, table_name):
        # Get the number of mysql table
        self.test_connection()
        sql = "select count(image_path) from " + table_name + ";"
        try:
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            LOGGER.debug(f"MYSQL count table:{table_name}")
            return results[0][0]
        except Exception as e:
            LOGGER.error(f"MYSQL ERROR: {e} with sql: {sql}")
            sys.exit(1)

4. 总结

本文使用图像保存例子介绍了如何使用pymysql库来对MySQL的数据表进行增删改查等操作，介绍pymysql的目的是：在后续利用Milvus进行以图搜图会涉及到使用MySQL来保存milvus的index索引对应的图像信息。这个也相当于是一些基础知识吧。