mysql-connector

MySQL-connector是一個Python模塊

1.安裝

python -m pip install mysql-connector

測試是否安裝成功

import mysql.connector

test數據庫結構

2.建立數據庫連接

db.py

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbprint(conn())

執行結果

python db.py<mysql.connector.connection.MySQLConnection object at 0x000001FF13E240F0>

3.數據插入

(1)單條插入

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef insert(name,now):    mydb = conn()    mycursor = mydb.cursor()    sql = "insert INTO test (name, create_time) VALUES (%s, %s)"    val = (name, now)    mycursor.execute(sql, val)    mydb.commit()    return mycursor.rowcountnow = int(time.time())print(insert('ABC',now))

執行結果

python db.py1

(2)批量插入

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef insertMore(val_list):    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "insert INTO {}(name, create_time)VALUES ({}, {});"    sql = ins_sql.format(table_name, '%s', '%s')    mycursor.executemany(sql, val_list)    mydb.commit()    return mycursor.rowcountname_list = ["AA", "BB", "CC", "DD"]now = int(time.time())val_list = [[name_list[i], now]            for i in range(len(name_list))]print(insertMore(val_list))

執行結果

python db.py4

4.數據查詢

(1)查詢部分字段的全部數據 fetchall()

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef findAll():    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "select name,from_unixtime(create_time,'%Y-%m-%d %H:%i:%s') FROM {}"    sql = ins_sql.format(table_name)    mycursor.execute(sql)    rs = mycursor.fetchall()     # 獲取所有記錄    return rsrs = findAll()for x in rs:  print(x)

執行結果

python db.py('ABC', '2022-09-16 11:17:32')('AA', '2022-09-16 11:18:06')('BB', '2022-09-16 11:18:06')('CC', '2022-09-16 11:18:06')('DD', '2022-09-16 11:18:06')

(2)查詢一條數據 fetchone()

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef findOne():    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "select name,from_unixtime(create_time,'%Y-%m-%d %H:%i:%s') FROM {}"    sql = ins_sql.format(table_name)    mycursor.execute(sql)    rs = mycursor.fetchone()    return rsrs = findOne()print(rs)

執行結果

python db.py('ABC', '2022-09-16 11:17:32')

(3)按條件獲取數據

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef findSome(val):    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "select name,from_unixtime(create_time,'%Y-%m-%d %H:%i:%s') FROM {} where name like '%{}%'"    sql = ins_sql.format(table_name,val)    mycursor.execute(sql)    rs = mycursor.fetchall()    return rsrs = findSome('A')for x in rs:  print(x)

執行結果

python db.py('ABC', '2022-09-16 11:17:32')('AA', '2022-09-16 11:18:06')

5.數據更新

(1)單條更新

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef update(val,id):    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "update {} SET name = {},update_time = {}  WHERE id = {}"    sql = ins_sql.format(table_name, '%s', '%s', id)    mycursor.execute(sql, val)    mydb.commit()    return mycursor.rowcountnow = int(time.time())val = ('XYZ', now)print(update(val,1))

執行結果

python db.py1

(2)批量更新

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef updateMore(val_list):    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "update {} SET name = {},update_time = {}  WHERE id = {}"    sql = ins_sql.format(table_name, '%s', '%s', '%s')    mycursor.executemany(sql, val_list)    mydb.commit()    return mycursor.rowcountids = ["2","3","4","5"]name_list = ["AA1", "BB1", "CC1", "DD1"]now = int(time.time())val_list = [(name_list[i], now,ids[i])            for i in range(len(name_list))]print(updateMore(val_list))

執行結果

python db.py4

6.數據刪除

import mysql.connectorimport time#數據庫連接def conn():    mydb = mysql.connector.connect(        host="localhost",        user="用戶名",        passwd="密碼",        database="數據庫名"    )    return mydbdef delect(val):    mydb = conn()    mycursor = mydb.cursor()    table_name = "test"    ins_sql = "delete FROM {} where name like '%{}%'"    sql = ins_sql.format(table_name, val)    mycursor.execute(sql)    mydb.commit()    return mycursor.rowcountprint(delect("A"))

執行結果

python db.py1