pymysql封装操作数据库

墨蓝 2022-04-13 15:20 324阅读 0赞

### pymysql封装操作数据库 ###

*  安装pymysql
 *  简单写操作
 *  简单读操作
 *  封装工具类多操作
 *  %d format: a number is required, not str 错误解决

# 安装pymysql #

#-i http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com 指定使用阿里云的源，可以不加
    pip3 install pymysql http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com

# 简单写操作 #

import pymysql.cursors
    
    # Connect to the database
    connection = pymysql.connect(host='localhost',
                                 user='user',
                                 password='passwd',
                                 db='db',
                                 charset='utf8mb4',
                                 cursorclass=pymysql.cursors.DictCursor)
    
    try:
        with connection.cursor() as cursor:
            # Create a new record
            sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
            cursor.execute(sql, ('webmaster@python.org', 'very-secret'))
    
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except Exception as e:
        print(e)
        # error rollback
        connection.rollback()
    finally:
        connection.close()

# 简单读操作 #

import pymysql.cursors
    
    # Connect to the database
    connection = pymysql.connect(host='localhost',
                                 user='user',
                                 password='passwd',
                                 db='db',
                                 charset='utf8mb4',
                                 cursorclass=pymysql.cursors.DictCursor)
    
    try:
        with connection.cursor() as cursor:
            # Read a single record
            sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
            cursor.execute(sql, ('webmaster@python.org',))
            result = cursor.fetchone()
            print(result)
    except Exception as e:
        print(e)
    finally:
        connection.close()

# 封装工具类多操作 #

[PyMysql.py][]

#! /usr/bin/env python3
    # -*- coding:utf-8 -*-
    
    import pymysql
    
    class PyMysql:
    
        host = 'localhost'
        user = 'user'
        password = 'pwssword'
        db = 'test'
        charset = 'utf8mb4'
        cursorclass = pymysql.cursors.DictCursor
    
        @classmethod
        def query(cls,sql,args=None,fetchone=False):
            # 创建连接
            connection = pymysql.connect(host=cls.host, user=cls.user,
                                         password=cls.password, db=cls.db, charset =cls.charset, cursorclass = cls.cursorclass)
            try:
                result = None
                # 开启游标
                with connection.cursor() as cursor:
                    # 返回响应结果数
                    effect_row = cursor.execute(cls.sql_args_2_sql(sql, args))
                    if fetchone:
                        result = cursor.fetchone()
                    else:
                        result = cursor.fetchall()
            except Exception as e:
                print(e)
            finally:
                # 关闭连接
                connection.close()
    
            return result
    
        @classmethod
        def execute(cls,sql,args=None,response=False):
            connection = pymysql.connect(host=cls.host, user=cls.user,
                                         password=cls.password, db=cls.db, charset =cls.charset, cursorclass = cls.cursorclass)
    
            try:
                result = None
                with connection.cursor() as cursor:
                    effect_row = cursor.execute(cls.sql_args_2_sql(sql, args))
                    if response:
                        result = cursor.fetchall()
    
                # connection is not autocommit by default. So you must commit to save your changes.
                connection.commit()
            except Exception as e:
                print(e)
                # error rollback
                connection.rollback()
            finally:
                connection.close()
    
            if response:
                return result
    
        @staticmethod
        def sql_args_2_sql(sql,args):
            ''' fix issue %d format: a number is required, not str :param sql: sql语句 :param args: 格式化参数 :return: 组合之后的sql语句 '''
            if args is None:
                return sql
            if sql.find('%') > -1:
                return sql % args
            elif sql.find('{') > -1:
                if type(args) is dict:
                    return sql.format(**args)
                else:
                    return sql.format(*args)
            return sql
    
    
    if __name__ == '__main__':
        #print(PyMysql.query(sql="select * from test where name = '%s' and age = %d ", args=('admin1', 18,), fetchone=True))
        #print(PyMysql.query(sql="select * from test where name = '{0}' and age = {1} ", args=('admin1', 18,),fetchone=True))
        #print(PyMysql.execute(sql="insert into test(name,gender,age) values('{name}','{gender}',{age})",args={'name':'admin6','gender':'女','age':18}))
        dict_result = PyMysql.query(sql="select * from test where name = '{0}' ", args=('admin1',),fetchone=True)
        print(dict_result)
        print(dict_result['name'],dict_result['age'],dict_result['gender'])
    
        PyMysql.execute(sql='UPDATE test set age = age +1 where id = %d ',args=(dict_result['id']))
    
        print(PyMysql.query(sql="select * from test where id = {0} ", args=(dict_result['id'],)))
    
        PyMysql.cursorclass = PyMysql.cursors.Cursor
        print(PyMysql.execute(sql="show tables",response=True))

# %d format: a number is required, not str 错误解决 #

@staticmethod
        def sql_args_2_sql(sql,args):
            ''' fix issue %d format: a number is required, not str :param sql: sql语句 :param args: 格式化参数 :return: 组合之后的sql语句 '''
            if args is None:
                return sql
            if sql.find('%') > -1:
                return sql % args
            elif sql.find('{') > -1:
                if type(args) is dict:
                    return sql.format(**args)
                else:
                    return sql.format(*args)
            return sql

[PyMysql.py]: http://PyMysql.py