用python開發(fā)一款操作MySQL的小工具

conn = MysqlConnection(string_arg="mysql -uroot -h127.0.0.1 -P3306 -p123456  -Dtest")
result,count=conn.execute_sql("select * from  test_table ")
print(result)
print(count)
>>> ((1, '羅輯', '28'), (2, '莊顏', '25'), (3, '葉文潔', '54'), (4, '程心', '25'), (5, '云天明', '27'))
>>> 5

 #這里默認(rèn)是普通游標(biāo)，你也可以指定使用字典游標(biāo)：

result, count = conn.execute_sql("select  * from  test_table ", cursor_type='dict')

>>>[{'ID': 1, 'name': '羅輯', 'age': '28'}, {'ID': 2, 'name': '莊顏', 'age': '25'}, {'ID': 3, 'name': '葉文潔', 'age': '54'}, {'ID': 4, 'name': '程心', 'age': '25'}, {'ID': 5, 'name': '云天明', 'age': '27'}]
>>>5

#傳元組
result,count=conn.execute_sql("select  * from  test_table where age=%s ",param=(25,))
#傳字典
result, count = conn.execute_sql("select  * from  test_table where age=%(age)s ", param={'age': 25})

#元組列表
result, count = conn.execute_sql("insert into  test_table(`age`,`name`)values(%s,%s) ", param=[('宋運(yùn)輝', 37), ('程開顏', 33)])

#字典列表
result, count = conn.execute_sql("insert into  test_table(`age`,`name`)values(%(age)s,%(name)s) ",
param=[ {"name":"宋運(yùn)輝",'age':37}, {"name":"程開顏",'age':33} ])

def select(self, columns='id', table=None, where=None, group_by: str = None, order_by: str = None, limit: int = None, offset=None,cursor_type=None):

def insert_into(self, table, data: dict or list, columns=None, ignroe=False, on_duplicate_key_update: str = None, return_id=False):

#傳入 字典
data_1 = {"age": 44, 'name': "雷東寶"}
count = conn.insert_into('test_table', data_1)#默認(rèn)返回受影響條數(shù)
print(count) #
>>> 1 
return_id = conn.insert_into('test_table', data_1,return_id=True)# 可選返回id
print(return_id)
>>>22533   

#傳入字典列表
data_2={"age": 22, 'name': "宋運(yùn)萍"}
all_data=[data_1,data_2]
count = conn.insert_into('test_table', all_data)

#限定 插入的字段。（字典有多字段，但是只需要往表里插入指定的字段時(shí)）
data_3= {"age": 44, 'name': "雷東寶","title":"村支書"} #title不需要,只要age和name
count = conn.insert_into('test_table', data_1,columns=["age","name"] )


#ignore 參數(shù)
data_1 = {"age": 44, 'name': "雷東寶","id":22539}
count = conn.insert_into('test_table',ignore=True )
print(count)
>>> 0   # 由于表中id 22539 已經(jīng)存在，該條記錄不會(huì)插入，影響 0條數(shù)據(jù)


#on_duplicate_key_update  參數(shù)
data_1 = {"age": 44, 'name': "雷東寶","id":22539} #id=22539 已經(jīng)存在
count = conn.insert_into('test_table', data_1,on_duplicate_key_update=' name="雷copy" ')
print(count)#返回影響條數(shù)
>>>2      #嘗試插入一條，但是發(fā)生重復(fù)，于是刪除新數(shù)據(jù)，并更新舊數(shù)據(jù)。實(shí)際上影響了兩條。

data_1 = {"age": 44, 'name': "雷東寶","id":22539}
data_2={"age": 22, 'name': "宋運(yùn)萍","id":22540}
all_data = [data_1, data_2,]
count=conn.merge_into('test_table',all_data,need_merge_columns=['name',])
print(count)
>>>4        #兩條數(shù)據(jù)正好都是重復(fù)的，插入兩條又刪除后修改兩條 ，返回4

data_1 = {"age": 44, 'name': "雷東寶","id":22539}
data_2={"age": 22, 'name': "宋運(yùn)萍","id":22540}
all_data = [data_1, data_2,]
count=conn.replace_into('test_table',all_data)

data_1 = {"age": 44, 'name': "雷copy"}
count=conn.update('test_table',data_1,where={'id':22539}) #更新 id=22539的數(shù)據(jù)為 新的data_1
print(count)
>>>1  

data_1 = {"age": 44, 'name': "雷cpy"}
count=conn.update_by_primary('test_table',data_1,pri_value=22539)

def delete_by_primary(self, table, pri_value, primary='id'):
	"""
	通過主鍵刪除數(shù)據(jù)
	"""

def delete(self,table, where: str or dict, limit: int = 0):
	"""
	通過where條件刪除數(shù)據(jù)
	"""


count=conn.delete('test_table',where={'name':'雷東寶'})          #刪除name=雷東寶的數(shù)據(jù)
count=conn.delete_by_primary('test_table',pri_value=22539)         #刪除主鍵等于22539 的數(shù)據(jù)

sql_params = [
        ("update test_table set name=%(name)s where  id=%(id)s ", {'name': '洛基', 'id': 22539}),
        ("update test_table set name=%(name)s where  id=%(id)s ", {'name': 'mask', 'id': 22540}),
    ]
count=conn.do_transaction(sql_params)
>>>((), 1)          #返回最后一條執(zhí)行語(yǔ)句的 結(jié)果和影響條數(shù)

result=conn.read_ss_result("select * from test_table")
for data in result:
	print(data)