Python 使用 PyMySQL 操作 Mysql 数据库

下载
连接数据库
- 使用 URI 创建连接
操作数据库
- 插入数据
- 查询数据
完整 demo

下载

1	$ sudo pip install PyMySQL

连接数据库

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author: wxnacy([email protected])

import pymysql.cursors
conn            = pymysql.connect(
    host        = 'localhost',
    port        = 3306,
    user        = 'root',
    password    = 'passwd'
    db          = 'your_db'
    charset     = 'utf8mb4',
    cursorclass = pymysql.cursors.DictCursor
)

使用 URI 创建连接

创建连接非常简单，但是配置在项目中却不方便，如果数据信息使用 URI 就简单了很多，这需要 urllib 模块的配合

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author: wxnacy([email protected])

import pymysql.cursors
from urllib.parse import urlparse

URI = 'mysql+pymysql://root:[email protected]:3306/study?charset=utf8mb4'
URL_CONFIG = urlparse(URI)

conn            = pymysql.connect(
    host        = URL_CONFIG.hostname,
    port        = URL_CONFIG.port,
    user        = URL_CONFIG.username,
    password    = URL_CONFIG.password,
    db          = URL_CONFIG.path[1:],
    charset     = 'utf8mb4',
    cursorclass = pymysql.cursors.DictCursor
)

注意地址模式需要使用 mysql+pymysql

操作数据库

插入数据

cursor = conn.cursor()
cursor.execute('insert into book (name) values (%s)', ['wxnacy'])
conn.commit()
cursor.close()

查询数据

cursor = conn.cursor()
cursor.execute('select * from book where name = %s', ['wxnacy'])
conn.commit()
res = cursor.fetchall()
cursor.close()
print(res)  # [{"id": 1, "name": "wxnacy"}]

完整 demo

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author: wxnacy([email protected])

import pymysql.cursors
from urllib.parse import urlparse

URI = 'mysql+pymysql://root:[email protected]:3306/study?charset=utf8mb4'
URL_CONFIG = urlparse(URI)
print(URL_CONFIG)

class BaseDB(object):
    @classmethod
    def create_conn(cls):
        '''创建mysql链接'''
        return pymysql.connect(
            host=URL_CONFIG.hostname,
            port=URL_CONFIG.port,
            user=URL_CONFIG.username,
            password=URL_CONFIG.password,
            db=URL_CONFIG.path[1:],
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor
        )

    @classmethod
    def query(cls, sql, params):
        """
        查询操作
        :param sql:
        :param params:
        :return:
        """
        conn = cls.create_conn()
        try:
            cursor = conn.cursor()

            cursor.execute(sql, params)
            conn.commit()
            result = cursor.fetchall()
            cursor.close()
            return result
        except BaseException as e:
            app.logger.error(traceback.format_exc())
            return []
        finally:
            conn.close()

    @classmethod
    def execute(cls, sql, params):
        """
        更新操作
        :param sql:
        :param params:
        :return:
        """
        conn = cls.create_conn()
        try:
            cursor = conn.cursor()

            result = cursor.execute(sql, params)
            conn.commit()
            cursor.close()
            return result
        except BaseException as e:
            app.logger.error(traceback.format_exc())
            return False
        finally:
            conn.close()

if __name__ == "__main__":
    sql = 'select * from user'
    res = BaseDB.query(sql, [])
    print(res)