def marketing_sql(sql_instruct: str = ""):
"""数据库添加身份认证"""
logs = log_message('marketing数据库')
try:
# 连接数据库
iden_sql = pymysql.connect(host="10.100.151.202",
port=4000,
user='u_pd_',账号
passwd="M6te",密码
db='123')库名
# 收发指令
cursor = iden_sql.cursor(cursor=pymysql.cursors.DictCursor)
time.sleep(1)
# 2.发送指令
sql = sql_instruct
cursor.execute(sql)
data_list = cursor.fetchone() # 查询一条数据，如果为空返回就是None
result = cursor.fetchall() # 记录查询结果，如果是空返回就是空元组（）
# # 存储本次指令查出的内容
# instruct_list = []
# for row_dict in data_list:
# instruct_list.append(row_dict)
# 3.关闭sql链接
cursor.close()
iden_sql.close()
return data_list
except Exception as e:
logs.logger.error('数据库异常,sql语句：{}，结果为:{}'.format(sql, str(e)))
return {'sql语句':sql,'exception':str(e)}1、通过python实现mysql数据库的查询。mysql数据库的查询可以分为两种，一种可以对数据库某个表内的数据进行查询，一种可以对数据表内的特定数据进行查询。
第一种：对数据表内的所有数据进行查询，查询显示结果是一个元组tuple。代码演示如下
def query1(sql):
"""
不带参数查询
:param sql:
:return:
"""
db = open()
cursor = db.cursor() # 使用cursor（）方法获取游标
cursor.execute(sql) # 执行sql查询语句
result = cursor.fetchall() # 记录查询结果
cursor.close() # 关闭游标
db.close() # 关闭数据库连接
return result # 返回查询结果
if __name__ == "__main__":
try:
sql2 = "select * from user_info"
tuple = query1(sql2)
print(tuple)
for index in range(len(tuple)):
temp_dict = dict()
temp_dict["userName"] = tuple[index][0]
temp_dict["userPwd"] = tuple[index][1]
all_user_info.append(temp_dict)
print(all_user_info)
except:
print("数据读取错误！")第二种：对数据内特定的数据进行查询。查询结果是元组tuple。代码演示如下：
def query(sql, *keys):
"""
带参数查询数据库数据
:return:
"""
db = open() # 连接数据库
cursor = db.cursor() # 使用cursor()方法获取操作游标
cursor.execute(sql, keys) # 执行查询sql 语句
result = cursor.fetchall() # 记录查询结果
cursor.close() # 关闭游标
db.close() # 关闭数据库连接
return result # 返回查询结果
if __name__ == "__main__":
try:
sql = "select * from user_info where userName=%s and userPwd=%s"
val = ("bob", "bob")
tuple = query(sql)
print(tuple)
for index in range(len(tuple)):
temp_dict = dict()
temp_dict["userName"] = tuple[index][0]
temp_dict["userPwd"] = tuple[index][1]
all_user_info.append(temp_dict)
print(all_user_info)
except:
print("数据读取错误！")2、通过python实现单条数据的添加。代码演示如下：
def insert(sql, values):
"""
向数据库插入数据,插入的values是一个tuple
:param sql: 运行sql插入语句
:return: 返回插入结果
"""
db = open() # 打开数据库连接
cursor = db.cursor() # 使用cursor（）方法获取游标
cursor.execute(sql, values)
db.commit()
cursor.close()
db.close()
return cursor.rowcount
if __name__ == "__main__":
try:
sql1 = "insert into user_info (userName,userPwd) values (%s,%s)"
val = ("bob", "bob")
print(insert(sql,val)，"条数据添加成功")
except:
print("数据操作错误！"）3、通过python实现多条数据批量添加到数据库中，使用executemany（）方法，添加的元素必须是一个list集合中的元素。代码演示如下：
def insert(sql, values):
"""
向数据库插入数据,插入的values是一个tuple
:param sql: 运行sql插入语句
:return: 返回插入结果
"""
db = open() # 打开数据库连接
cursor = db.cursor() # 使用cursor（）方法获取游标
cursor.executemany(sql, values)
db.commit()
cursor.close()
db.close()
return cursor.rowcount
if __name__ == "__main__":
try:
sql1 = "insert into user_info (userName,userPwd) values (%s,%s)"
val = [("bob", "bob"),("tom", "tom")]
print(insert(sql,val)，"条数据添加成功")
except:
print("数据操作错误！"）4、通过python实现mysql数据数据的删除。代码演示如下
def delete(sql, values):
"""
删除数据库数据
:param sql:
:return:
"""
db = open() # 打开数据库连接
cursor = db.cursor() # 使用cursor()方法获取游标
cursor.execute(sql, values)
db.commit() # 执行修改
cursor.close()
db.close()
return cursor.rowcount
if __name__ == "__main__":
try:
sql_del = "delete from user_info where userName = %s"
keys = "test"
print(delete(sql_del, keys),"条数据删除成功！")
except:
print("数据读取错误！")5、通过python实现数据库库表内数据更新。代码演示如下：
def update(sql, values):
"""
更新数据库数据
:param sql:
:param values:
:return:
"""
db = open() # 打开数据库连接
cursor = db.cursor() # 使用cursor()方法获取游标
cursor.execute(sql, values) # 执行sql数据修改语句
db.commit() # 提交数据
cursor.close() # 关闭游标
db.close() # 关闭数据库连接
return cursor.rowcount
if __name__ == "__main__":
try:
sql_update = "update user_info set userName = %s where userPwd=%s"
val = ('test', '123456')
print(update(sql_update, val), "条数据被修改！")
except:
print("数据读取错误！")