PyMySQL

PyMySQL It's from Python Connect to MySQL Database server interface . It has achieved Python database API v2.0, And contains a pure Python Of MySQL Client library

install PyMySQL

pip install PyMySQL
 If the download speed is slow , Domestic image source can be used 
pip install PyMySQL -i https://pypi.tuna.tsinghua.edu.cn/simple

After successful installation , Test whether you can connect to the database

import pymysql

#  Create links , Connect to database （ Server address 、 Port number 、 user name 、 password 、 Database name 、 Coding format ）
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Use cursor() Method to get a cursor ( The query data is returned in tuple format )
cursor = conn.cursor()

#  perform sql Method of querying version information in statement 
sql = '''SELECT VERSION(); '''
cursor.execute(sql)

#  Query all the data , The returned data is in tuple format 
data = cursor.fetchall()

#  Close cursor 
cursor.close()

#  Close links 
conn.close()

print(" The database version is :{}".format(data))
 Print the results 
 The database version is :(('10.3.27-MariaDB-0+deb10u1',),)

【 Be careful 】：sql It's best to end the sentence with ;

Create database tables

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Create cursors 
cursor = conn.cursor()

#  If the table exists, delete 
cursor.execute("DROP TABLE IF EXISTS users")

#  Prepare statements for creating tables 
sql = """CREATE TABLE users( id int(10) NOT NULL AUTO_INCREMENT, phonenum int(20) NOT NULL, nickname char(20) DEFAULT NULL, sex char(1) DEFAULT NULL, birth_year int(4) DEFAULT NULL, birth_month int(2) DEFAULT NULL, birth_day int(2) DEFAULT NULL, avatar char(10) DEFAULT NULL, location char(10) DEFAULT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;"""
# perform sql sentence 
cursor.execute(sql)


#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

The insert

When a record is created into a database table , You need to perform INSERT operation .

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Create cursors 
cursor = conn.cursor()

#  Get ready sql sentence 
sql = """INSERT INTO users(phonenum,nickname, sex, birth_year,birth_month,birth_day,avatar,location) VALUES (13766665555,'bobo',1,1998,5,27,'00','gz')"""
try:
   #  perform sql sentence 
   cursor.execute(sql)
   #  Commit transaction 
   conn.commit()
except:
   #  If an error occurs , Then roll back 
   conn.rollback()


#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

Query operation

After establishing the database connection , You can query this database . have access to fetchone() Method to get a single record or fetchall() Method to get multiple values from the database table .

• fetchone() Get the next row of the query result set . The result set is the object returned when the current cursor object is used to query the table .

• fetchmany(n) Get the next... Of the query result set n That's ok . The result set is the object returned when the current cursor object is used to query the table .

• fetchall() Get all the rows in the result set . If you have extracted some rows from the result set , Then the remaining rows are retrieved from the result set .

users Table content

The cursor is in the initial position , It can be done by fetchall() Get all the rows in the result set

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Create cursors ( The query data is returned in tuple format )
cursor = conn.cursor()

# 1.  perform SQL, Query operation 
sql1= '''SELECT * FROM users WHERE sex =1 ; '''

effect_row1 = cursor.execute(sql1)

# 2.  Show all the data queried , The returned data is in tuple format 
result1 = cursor.fetchall()

#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

print(result1)

#  Print the results 
''' ( (1, '13766665555', 'bobo', 1, 1998, 5, 27, '00', 'gz'), (3, '13744443333', 'mingming', 1, 1999, 2, 3, '02', 'sh'), (5, '13722221111', 'hanghang', 1, 1999, 4, 4, '00', 'gz'), (6, '17311112222', 'zhanzhan', 1, 2000, 3, 3, '02', 'sh') ) '''

The cursor is in the initial position , It can be done by fetchone() Get the first row of the query result set

（ For data in the database , After the query, in order to facilitate processing , Can be converted to dictionary format , adopt key-value Get the data you need ）

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')


#  Create cursors ( The query data is returned in dictionary format )
cursor = conn.cursor(pymysql.cursors.DictCursor)

# 1.  perform SQL, Query operation 
sql1= '''SELECT * FROM users WHERE sex =1 ; '''

effect_row1 = cursor.execute(sql1)

# 2.  Only the first item of query data is displayed 
result1 = cursor.fetchone()

#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

print(result1)

#  Print the results 
''' {'id': 1, 'phonenum': '13766665555', 'nickname': 'bobo', 'sex': 1, 'birth_year': 1998, 'birth_month': 5, 'birth_day': 27, 'avatar': '00', 'location': 'gz'} '''

adopt fetchone() When getting data , The cursor is in the initial position , therefore fetchone() The obtained data is the first data in the query set ;

adopt fetchmany(2) When getting data , The cursor is on the next line of the initial position , therefore fetchmany(2) The obtained data is the second and third data in the query set ;

adopt fetchall() When getting data , The cursor also moves , At this time, there is only one piece of data left in the test set , therefore fetchall() The obtained data is the last data in the query set （ Article 4. ）;

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')


#  Create cursors ( The query data is returned in tuple format )
cursor = conn.cursor()

# 1.  perform SQL
sql1= '''SELECT * FROM users WHERE sex =1 ; '''

effect_row1 = cursor.execute(sql1)

# 2.  Return the data 
result1 = cursor.fetchone()
result2 = cursor.fetchmany(2)
result3 = cursor.fetchall()

#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

print(result1)
print(result2)
print(result3)

#  Print the results 
''' result1 (1, '13766665555', 'bobo', 1, 1998, 5, 27, '00', 'gz') result2 ( (3, '13744443333', 'mingming', 1, 1999, 2, 3, '02', 'sh'), (5, '13722221111', 'hanghang', 1, 1999, 4, 4, '00', 'gz') ) result3 ( (6, '17311112222', 'zhanzhan', 1, 2000, 3, 3, '02', 'sh'), ) '''

【 expand 】 ： perform sql When the sentence is ,sql The parameters inside , Dynamic parameter transfer may be required ,
such as select * from users where phonenum ='13766665555'; The phone number inside , Need from excel Get it dynamically in the file, and then query , There are two ways .

• One is to adopt format function ：

sql = "select * from users where phonenum ={};".format(phone)

But in general, it is not recommended to use format function , because format The content in the function can be relatively long , More content , May cause sql A series of security issues such as injection , For example, insert and execute a delete statement , It may cause the deletion of the table

• Two is through %s Place holder
  Hold down Ctrl, Click on cursor.execute(sql1) Medium execute() function , Jump to the source code

def execute(self, query, args=None):
     """Execute a query :param str query: Query to execute. :param args: parameters used with query. (optional) :type args: tuple, list or dict :return: Number of affected rows :rtype: int If args is a list or tuple, %s can be used as a placeholder in the query. If args is a dict, %(name)s can be used as a placeholder in the query. """
     while self.nextset():
         pass

     query = self.mogrify(query, args)

     result = self._query(query)
     self._executed = query
     return result

This explains , execute() Functions can pass arguments , This parameter can be a list 、 Tuples 、 Dictionaries

sql = "select * FROM users WHERE birth_year < %s and birth_month > %s "

cursor.execute(sql=sql,args=[2000,4])

It's a safer way

Encapsulating queries

For data in the database , Query operation is the most common operation , In the interface automation test project , You often have to query the data in the database , Here is a simple encapsulation of the query operation

Here are two packaging methods
The first one is ： The database configuration does not need to be changed , Then fill in the database information into the initialization configuration

import pymysql
from pymysql.cursors import DictCursor


class MysqlUtil:

    def __init__(self,return_dict=False):
        #  Configuration database 
        self.coon = pymysql.connect(
            host="192.168.174.132",
            port=3306,
            user='bobo',
            passwd='hh123456',
            db='swiper',
            charset='utf8'
        )
        # self.cursor = self.coon.cursor(DictCursor)
        if return_dict:
            self.cursor = self.coon.cursor(pymysql.cursors.DictCursor)  #  Specifies that each row of data is returned as a dictionary 
        else:
            self.cursor = self.coon.cursor()  #  Specifies that each row of data is returned as a primitive 

    def close(self):
        self.cursor.close()  #  Close the query 
        self.coon.close()  #  Close the connection 

	#  Query a piece of data 
    def fetch_one(self, sql):
        #  perform SQL
        self.cursor.execute(sql)
        #  To get the results 
        result = self.cursor.fetchone()  #  Return to Yuanzu （）/  Return dictionary  {}
        return result  #  Return results 

	#  Query multiple data 
    def fetch_all(self, sql):
        #  perform SQL
        self.cursor.execute(sql)
        #  To get the results 
        results = self.cursor.fetchall()  #  Returns a list of  [(),()...] / [{},{}...]
        return results




if __name__ == '__main__':
    mysql = MysqlUtil(return_dict=True)
    sql = "select * from users where phonenum ='13766665555';"
    result = mysql.fetch_one(sql)
    print(result["birth_day"])
    #  Be sure to turn it off 
    mysql.close()

#  Print the results  27 

The second kind ： For safety reasons , Write the database information to the configuration file , Classes are more reusable , And sql More secure

import pymysql
from pymysql.cursors import DictCursor

class DBHandler:
	# init The data in is subsequently replaced by config File read 
    def __init__(self,host="192.168.174.132",
                 user='bobo',
                 passwd='hh123456',
                 db='swiper',
                 charset='utf8',
                 cursorclass=DictCursor,
                 **kw
                     ):

            #  Create connection 
            self.con = pymysql.connect(host=host,user=user,passwd=passwd,db=db,charset=charset,cursorclass=cursorclass,**kw)
            #  Create cursors ( The query data is returned in dictionary format )
            self.cursor = self.con.cursor()

    def query(self,sql,args=None,query_one=True):
        ''' Query statement , By default, only one query '''
        self.cursor.execute(sql,args)
        #  To get the results 
        if query_one :
            return self.cursor.fetchone()
        else:
            return self.cursor.fetchall()

    def close(self):
        self.cursor.close()  #  Close the query 
        self.con.close()  #  Close the connection 


if __name__ == '__main__':
    sql = "select * FROM users WHERE birth_year < %s and birth_month>%s "
    db = DBHandler()
    res =db.query(sql=sql,args=[2000,4],one=False)
    print(res)

#  result 
[
{
    'id': 1, 'phonenum': '13766665555', 'nickname': 'bobo', 'sex': 1, 'birth_year': 1998, 'birth_month': 5, 'birth_day': 27, 'avatar': '00', 'location': 'gz'},
{
    'id': 2, 'phonenum': '13755554444', 'nickname': 'yiyi', 'sex': 0, 'birth_year': 1999, 'birth_month': 5, 'birth_day': 5, 'avatar': '00', 'location': 'hz'}
]

update operation

UPDATE Statement can update the data in any database , It can be used to update one or more existing records in the database .

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Create cursors 
cursor = conn.cursor()

#  Get ready sql sentence （ take sex=0 Year of birth +1）
sql = "UPDATE users SET birth_year = birth_year + 1 WHERE SEX = '%c'" % ('0')

try:
   #  perform sql sentence 
   cursor.execute(sql)
   #  Commit transaction 
   conn.commit()
except:
   #  If an error occurs , Then roll back 
   conn.rollback()


#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

Delete operation

When you want to delete some records from the database , Then you can execute DELETE operation .

import pymysql

#  Create connection 
conn = pymysql.connect(host="192.168.134.1",
                       port=3306,
                       user='root',
                       passwd='123456',
                       db='testdb',
                       charset='utf8')

#  Create cursors 
cursor = conn.cursor()

#  Get ready sql sentence （ take birth_year  stay 2000 People deleted after years ）
sql = "DELETE FROM users WHERE birth_year > '%d'" % (2000)

try:
   #  perform sql sentence 
   cursor.execute(sql)
   #  Commit transaction 
   conn.commit()
except:
   #  If an error occurs , Then roll back 
   conn.rollback()


#  Close cursor 
cursor.close()

#  Close the connection 
conn.close()

encapsulation PyMySQL

dbconf.yaml

db_product:
  dbname: test
  host: test.com
  port: 3306
  user: Admin
  pwd: 123456

db_test:
  dbname: testdb
  host: 192.168.132.1
  port: 3306
  user: bobo
  pwd: 123456

By reading the yaml Configuration information in the file , Connect to database , Perform database operations

DBHandler.py

import pymysql
from pymysql.cursors import DictCursor



class MysqlUtil:

    def __init__(self,dbconf):
        self.dbconf = dbconf
        self.conn = self.get_conn()  #  Connection object 
        self.cursor = self.get_cursor()  #  Cursor object 

    def get_conn(self):
        """  Get the connection object  """
        conn = pymysql.connect(host=self.dbconf['host'],
                               port=self.dbconf['port'],
                               user=self.dbconf['user'],
                               passwd=self.dbconf['pwd'],
                               db=self.dbconf['dbname'],
                               charset='utf8')
        return conn

    def get_cursor(self):
        """ Get cursor object """
        # cursor = None
        cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
        return cursor

    def query(self, sql, args=None, one=True):
        ''' Query statement , By default, only one query '''
        self.cursor.execute(sql, args)
        #  To get the results 
        if one:
            return self.cursor.fetchone()
        else:
            return self.cursor.fetchall()

    def commit_data(self, sql):
        """  Submit data ( to update 、 Insert 、 Delete operation ) """
        self.cursor.execute(sql)
        self.conn.commit()

    def close(self):
        self.cursor.close()
        self.conn.close()

YamlHandler.py

import yaml

#  Read yaml file 
class ReadYaml:
    def __init__(self, path, param=None):
        self.path = path  #  File path 
        self.param = param  #  Get all data by default 

    #  obtain yaml The data in the file 
    def get_data(self, encoding='utf-8'):
        with open(self.path, encoding=encoding) as f:
            data = yaml.load(f.read(), Loader=yaml.FullLoader)

            if self.param == None:
                return data  #  Return all data 
            else:
                return data.get(self.param)  #  The get key is param Value 

Test execution file

test.py

from DBHandler import MysqlUtil
from YamlHandler import ReadYaml
import os


if __name__ == '__main__':
	dir_path = os.path.split(os.path.abspath(__file__))[0]
    yaml_path = os.path.join(dir_path,'test.yaml')
    db_conf = ReadYaml(yaml_path,'db_test').get_data()
	
	db = MysqlUtil(db_conf)
    sql = "select * FROM users WHERE birth_year < %s and birth_month>%s "
    res = db.query(sql=sql, args=[2000, 4], one=False)
    print(res)
    db.close()