【pymysql】基於pymysql封裝的MySQL增刪改查操作

阿新 • • 發佈：2022-03-16

1. 官網

https://pymysql.readthedocs.io/en/latest/

2. 安裝

You can install it with pip:

$ python3 -m pip install PyMySQL
To use "sha256_password" or "caching_sha2_password" for authenticate, you need to install additional dependency:

$ python3 -m pip install PyMySQL[rsa]
To use MariaDB's "ed25519" authentication method, you need to install additional dependency: 


$ python3 -m pip install PyMySQL[ed25519]

3. 官方示例

import pymysql.cursors

# Connect to the database
connection = pymysql.connect(host='localhost',
                             user='user',
                             password='passwd',
                             database='db',
                             cursorclass 
=pymysql.cursors.DictCursor)

with connection:
    with connection.cursor() as cursor:
        # Create a new record
        sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
        cursor.execute(sql, ('[email protected]', 'very-secret'))

    # connection is not autocommit by default. So you must commit to save 

    # your changes.
    connection.commit()

    with connection.cursor() as cursor:
        # Read a single record
        sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
        cursor.execute(sql, ('[email protected]',))
        result = cursor.fetchone()
        print(result)

4. 關於cursor

1. 最常用的也是預設的遊標就是cursor，返回的資料格式為tuple

2. 其餘的遊標型別還有DictCursor，SSCursor，SSDictCursor等，SS開頭的遊標稱為流式遊標，

3. Cursor和DictCursor遊標可以一次性返回所有的資料，

4. 流式遊標智慧一條一條得返回查詢資料，所以這類遊標適用於記憶體低、網路頻寬小、資料量大的應用場景中。

DictCursor：返回字典（Dict）格式的資料

SSCursor：流式遊標返回元組（Tuple）格式資料

SSDictCursor：流式遊標返回字典（Dict）格式資料

5. github

https://github.com/PyMySQL/PyMySQL

6. 封裝

#!/usr/bin/env python  
# -*- coding:utf-8 -*-  

import pymysql
from pymysql import cursors

from common.framework.utils import debug_logger
from conf import config


class MysqlConnectCom(object):

    def __init__(self, db, user, passwd, host='localhost', port=3306, charset='utf8'):
        self.db = db
        self.user = user
        self.passwd = passwd
        self.host = host
        self.port = port
        self.charset = charset
        self.connect = None
        self.cursor = None

    def _connect_db(self):
        params = {
            "db": self.db,
            "user": self.user,
            "passwd": self.passwd,
            "host": self.host,
            "port": self.port,
            "charset": self.charset
        }
        self.connect = pymysql.connect(**params)
        self.cursor = self.connect.cursor(cursors.DictCursor)

    def _close_db(self):
        """
        關閉資料庫
        """
        self.cursor.close()
        self.connect.close()

    def select_operation(self, sql):
        if not sql.upper().startswith('SELECT'):
            debug_logger.error(f'[mysql] sql not startswith SELECT')
            return Exception('sql not right!')
        try:
            self.cursor.execute(sql)
        except Exception as e:
            debug_logger.error(f'[mysql]{e}')
            return []
        else:
            return self.cursor.fetchall()
        finally:
            self._connect_db()

    def insert_operation(self, sql):
        if not sql.upper().startswith('INSERT'):
            debug_logger.error(f'[mysql] sql not startswith INSERT')
            return Exception('sql not right!')
        try:
            count = self.cursor.execute(sql)
            self.connect.commit()
        except Exception as e:
            debug_logger.error(f'[mysql]{e}')
            self.connect.rollback()
            return False
        else:
            return count
        finally:
            self._connect_db()

    def update_operation(self, sql):
        if not sql.upper().startswith('UPDATE'):
            debug_logger.error(f'[mysql] sql not startswith UPDATE')
            return Exception('sql not right!')
        try:
            count = self.cursor.execute(sql)
            self.connect.commit()
        except Exception as e:
            debug_logger.error(f'[mysql]{e}')
            self.connect.rollback()
            return False
        else:
            return count
        finally:
            self._connect_db()

    def delete_operation(self, sql):
        if not sql.upper().startswith('DELETE'):
            debug_logger.error(f'[mysql] sql not startswith DELETE')
            return Exception('sql not right!')
        try:
            count = self.cursor.execute(sql)
            self.connect.commit()
        except Exception as e:
            debug_logger.error(f'[mysql]{e}')
            self.connect.rollback()
            return False
        else:
            return count
        finally:
            self._connect_db()


if __name__ == "__main__":
    client = MysqlConnectCom(config.DATABASE , config.USER,  config.PASSWORD, host=config.HOST)
    client._connect_db()
    # print(client.select_operation("SELECT * FROM student;"))