Python pymysql Operation MySQL Details

  • 2021-12-04 10:41:18
  • OfStack

Directory 1, use 1.1 simple use
1.2 Encapsulation tool classes

1. Use

1.1 Easy to use


import pymysql

#  Create a connection 
con = pymysql.connect(
                        host='localhost',
                        port=3306,
                        user='root',
                        password='123456',
                        database='test',
                        charset='utf8'
)
#  Create a cursor 
cursor = con.cursor()

#  Perform new SQL Returns the number of rows affected 
row1 = cursor.execute("insert into user (username, password) values ('username3','password3')")
print(row1)

#  Perform an update SQL Returns the number of rows affected 
row2 = cursor.execute("update user set password = '123456' where id > 2;")
#  Execute a query SQL
res = cursor.execute("SELECT * FROM user;")
result = cursor.fetchall()
for info in result:
    print(info[0], info[1])
#  Submit, otherwise new or updated data cannot be saved 
con.commit()
#  Close the cursor 
cursor.close()
#  Close the connection 
con.close()

Note: When Chinese exists in the database table, creating a connection requires specifying charset='utf8' Otherwise, the Chinese display will be garbled. Among them cursor.fetchall() Is to get all result sets. If there is only one result set, you can use the cursor.fetchone() .

1.2 Encapsulation tool classes

For ease of use, it can be directly encapsulated into tool classes:


import pymysql


class MysqlHelper:
    def __init__(self, config):
        self.host = config["host"]
        self.port = config["port"]
        self.user = config["user"]
        self.password = config["password"]
        self.db = config["db"]
        self.charset = config["charset"]
        self.con = None
        self.cursor = None

    def create_con(self):
        """
         Create a connection 
        """
        try:
            self.con = pymysql.connect(host=self.host, port=self.port, user=self.user, password=self.password,
                                       database=self.db, charset='utf8')
            self.cursor = self.con.cursor()
            return True
        except Exception as e:
            print(e)
            return False

    #
    def close_con(self):
        """
         Close a link 
        """
        if self.cursor:
            self.cursor.close()
        if self.con:
            self.con.close()

    # sql Execute 
    def execute_sql(self, sql):
        """
         Perform an insert / Update / Delete statement 
        """
        try:
            self.create_con()
            print(sql)
            self.cursor.execute(sql)
            self.con.commit()
        except Exception as e:
            print(e)
        finally:
            self.close_con()

    def select(self, sql, *args):
        """
         Execute the query statement 
        """
        try:
            self.create_con()
            print(sql)
            self.cursor.execute(sql, args)
            res = self.cursor.fetchall()
            return res
        except Exception as e:
            print(e)
            return False
        finally:
            self.close_con()

Using tool classes:


config = {
    "host": 'localhost',
    "port": 3306,
    "user": 'root',
    "password": '123456',
    "db": 'test',
    "charset": 'utf8'
}
db = MysqlHelper(config)
db.execute_sql("insert into user (username, password) values ('username4','password4')")
db.select("SELECT * FROM user;")

Related articles: