54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
from django.db import connection
|
|
|
|
def execute_raw_sql(sql: str, params=None, return_type: int =1):
|
|
"""执行原始sql并返回数据
|
|
|
|
Args:
|
|
sql (str): 查询语句
|
|
params (_type_, optional): 参数列表. Defaults to None.
|
|
return_type (int, optional): 返回格式. Defaults to 1.
|
|
1 直接返回包含多个字典的列表或空列表
|
|
2 返回原始数据
|
|
"""
|
|
with connection.cursor() as cursor:
|
|
if params:
|
|
cursor.execute(sql, params=params)
|
|
else:
|
|
cursor.execute(sql)
|
|
columns = [desc[0] for desc in cursor.description]
|
|
rows = cursor.fetchall()
|
|
if return_type == 1:
|
|
return [dict(zip(columns, row)) for row in rows]
|
|
else:
|
|
return rows
|
|
|
|
def query_all_dict(sql, params=None):
|
|
'''
|
|
查询所有结果返回字典类型数据
|
|
:param sql:
|
|
:param params:
|
|
:return:
|
|
'''
|
|
with connection.cursor() as cursor:
|
|
if params:
|
|
cursor.execute(sql, params=params)
|
|
else:
|
|
cursor.execute(sql)
|
|
columns = [desc[0] for desc in cursor.description]
|
|
return [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
def query_one_dict(sql, params=None):
|
|
"""
|
|
查询一个结果返回字典类型数据
|
|
:param sql:
|
|
:param params:
|
|
:return:
|
|
"""
|
|
with connection.cursor() as cursor:
|
|
if params:
|
|
cursor.execute(sql, params=params)
|
|
else:
|
|
cursor.execute(sql)
|
|
columns = [desc[0] for desc in cursor.description]
|
|
row = cursor.fetchone()
|
|
return dict(zip(columns, row)) |