68 lines
1.6 KiB
Python
68 lines
1.6 KiB
Python
|
# coding=utf-8
|
|||
|
|
"""
|
|||
|
|
@project: maxkb
|
|||
|
|
@Author：虎
|
|||
|
|
@file： sql_execute.py
|
|||
|
|
@date：2023/9/25 20:05
|
|||
|
|
@desc:
|
|||
|
|
"""
|
|||
|
|
from typing import List
|
|||
|
|
|
|||
|
|
from django.db import connection
|
|||
|
|
|
|||
|
|
|
|||
|
|
def sql_execute(sql: str, params):
|
|||
|
|
"""
|
|||
|
|
执行一条sql
|
|||
|
|
:param sql: 需要执行的sql
|
|||
|
|
:param params: sql参数
|
|||
|
|
:return: 执行结果
|
|||
|
|
"""
|
|||
|
|
with connection.cursor() as cursor:
|
|||
|
|
cursor.execute(sql, params)
|
|||
|
|
columns = list(map(lambda d: d.name, cursor.description))
|
|||
|
|
res = cursor.fetchall()
|
|||
|
|
result = list(map(lambda row: dict(list(zip(columns, row))), res))
|
|||
|
|
cursor.close()
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
|
|||
|
|
def update_execute(sql: str, params):
|
|||
|
|
"""
|
|||
|
|
执行一条sql
|
|||
|
|
:param sql: 需要执行的sql
|
|||
|
|
:param params: sql参数
|
|||
|
|
:return: 执行结果
|
|||
|
|
"""
|
|||
|
|
with connection.cursor() as cursor:
|
|||
|
|
cursor.execute(sql, params)
|
|||
|
|
affected_rows = cursor.rowcount
|
|||
|
|
cursor.close()
|
|||
|
|
return affected_rows
|
|||
|
|
|
|||
|
|
|
|||
|
|
def select_list(sql: str, params: List):
|
|||
|
|
"""
|
|||
|
|
执行sql 查询列表数据
|
|||
|
|
:param sql: 需要执行的sql
|
|||
|
|
:param params: sql的参数
|
|||
|
|
:return: 查询结果
|
|||
|
|
"""
|
|||
|
|
result_list = sql_execute(sql, params)
|
|||
|
|
if result_list is None:
|
|||
|
|
return []
|
|||
|
|
return result_list
|
|||
|
|
|
|||
|
|
|
|||
|
|
def select_one(sql: str, params: List):
|
|||
|
|
"""
|
|||
|
|
执行sql 查询一条数据
|
|||
|
|
:param sql: 需要执行的sql
|
|||
|
|
:param params: 参数
|
|||
|
|
:return: 查询结果
|
|||
|
|
"""
|
|||
|
|
result_list = sql_execute(sql, params)
|
|||
|
|
if result_list is None or len(result_list) == 0:
|
|||
|
|
return None
|
|||
|
|
return result_list[0]
|