import pymysql
class OperationMysql:
"""
? ? 數(shù)據(jù)庫SQL相關(guān)操作import pymysql
# 打開數(shù)據(jù)庫連接db = pymysql.connect("localhost","testuser","test123","TESTDB" )
# 使用 cursor() 方法創(chuàng)建一個游標(biāo)對象cursor
cursor = db.cursor()
# 使用 execute()? 方法執(zhí)行 SQL 查詢cursor.execute("SELECT VERSION()")
"""
? ? def __init__(self):
# 創(chuàng)建一個連接數(shù)據(jù)庫的對象
? ? ? ? self.conn = pymysql.connect(
host='00000000',? # 連接的數(shù)據(jù)庫服務(wù)器主機(jī)名
? ? ? ? ? ? port=3306,? # 數(shù)據(jù)庫端口號
? ? ? ? ? ? user='paidan_user',? # 數(shù)據(jù)庫登錄用戶名
? ? ? ? ? ? passwd='aaA5y6C9vL',
? ? ? ? ? ? db='test2',? # 數(shù)據(jù)庫名稱
? ? ? ? ? ? charset='utf8',? # 連接編碼
? ? ? ? ? ? cursorclass=pymysql.cursors.DictCursor
)
# 使用cursor()方法創(chuàng)建一個游標(biāo)對象，用于操作數(shù)據(jù)庫
? ? ? ? self.cur =self.conn.cursor()
# 查詢一條數(shù)據(jù)
? ? def search_one(self, sql):
self.cur.execute(sql)
result =self.cur.fetchone()# 使用 fetchone()方法獲取單條數(shù)據(jù).只顯示一行結(jié)果
? ? ? ? # result = self.cur.fetchall()? # 顯示所有結(jié)果
? ? ? ? return result
# 更新SQL
? ? def updata_one(self, sql):
try:
self.cur.execute(sql)# 執(zhí)行sql
? ? ? ? ? ? self.conn.commit()# 增刪改操作完數(shù)據(jù)庫后，需要執(zhí)行提交操作
? ? ? ? except:
# 發(fā)生錯誤時回滾
? ? ? ? ? ? self.conn.rollback()
self.conn.close()# 記得關(guān)閉數(shù)據(jù)庫連接
? ? # 插入SQL
? ? def insert_one(self, sql):
try:
self.cur.execute(sql)# 執(zhí)行sql
? ? ? ? ? ? self.conn.commit()# 增刪改操作完數(shù)據(jù)庫后，需要執(zhí)行提交操作
? ? ? ? except:
# 發(fā)生錯誤時回滾
? ? ? ? ? ? self.conn.rollback()
self.conn.close()
# 刪除sql
? ? def delete_one(self, sql):
try:
self.cur.execute(sql)# 執(zhí)行sql
? ? ? ? ? ? self.conn.commit()# 增刪改操作完數(shù)據(jù)庫后，需要執(zhí)行提交操作
? ? ? ? except:
# 發(fā)生錯誤時回滾
? ? ? ? ? ? self.conn.rollback()
self.conn.close()
if __name__ =='__main__':
op_mysql = OperationMysql()
res = op_mysql.search_one("SELECT * FROM order_case where id = '612244' ")
print(res)
from test_case.test_2import OperationMysql? ?導(dǎo)入方法
sql=OperationMysql().select_one('select * from? application ')
