一、开发环境搭建和配置
pip install mysql-connector-python
pip install PyMySQL
二、django模块目录
dbOperations
├── __init__.py
├── __pycache__
│ ├── __init__.cpython-313.pyc
│ ├── admin.cpython-313.pyc
│ ├── apps.cpython-313.pyc
│ ├── models.cpython-313.pyc
│ └── views.cpython-313.pyc
├── admin.py
├── apps.py
├── migrations
│ ├── __init__.py
│ └── __pycache__
│ └── __init__.cpython-313.pyc
├── models.py
├── tests.py
└── views.py三、django模块相关代码
***************************settings.py*********************************************
#数据库 连接配置 mysql
DATABASES = {
'default': {
# 数据库引擎
'ENGINE': 'django.db.backends.mysql',
# 'ENGINE': 'sqlalchemy.create_engine',
# 数据库名字
'NAME': 'xx',
# 数据库用户名
'USER': 'xx',
# 数据库密码
'PASSWORD': 'xx',
# 数据库主机地址
'HOST': 'xx',
# 数据库端口号
'PORT': 'xx'
}
}************************************urls************************************
urlpatterns = [
path('dbOperations/executeSQLView', db_views.executeSQLView,name='execute_sql_view'),
]**************************************viwes.py************************************
import json
import logging
from django.db import connection
from django.http import JsonResponse
from django.shortcuts import render
from django.views.decorators.csrf import csrf_exempt
from rest_framework import status
from operation.common.enum.ResponeCodeEnum import ResponseCodeEnum
from operation.common.exception.BusinessException import BusinessException
from operation.common.utils.CommonResult import CommonResult
from operation.common.utils.Serializers import SQLExecutionSerializer
# Create your views here.
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.db import connection
import logging
from rest_framework import status
import json
@csrf_exempt
def executeSQLView(request):
try:
if request.method == 'POST':
json_data = request.body
print(json_data)
data = json.loads(json_data)
sql = data.get('sql', None)
# 序列化器的正确使用：在 serializer = SQLExecutionSerializer(data={'sql': sql}) 中，
# 我们传递了一个包含 sql 键的字典，而不是直接传递 SQL 查询字符串。
serializer = SQLExecutionSerializer(data={'sql': sql})
if serializer.is_valid():
sql = serializer.validated_data['sql']
logging.info("sql:%s" % (sql))
"""
使用 with 语句管理游标
"""
with connection.cursor() as cursor:
try:
# 执行 SQL 查询：
cursor.execute(sql)
results = cursor.fetchall()
# 处理查询结果：
columns = [col[0] for col in cursor.description]
# 将查询结果转换为字典列表，每个字典表示一行数据。
data = [dict(zip(columns, row)) for row in results]
return JsonResponse(CommonResult.success_data(data), json_dumps_params={'ensure_ascii': False})
except Exception as e:
# return JsonResponse(CommonResult.error(e.code, e.message),
# json_dumps_params={'ensure_ascii': False})
raise BusinessException(e.message, e.code)
return JsonResponse(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
else:
raise BusinessException(ResponseCodeEnum.METHOD_ERROR.message, ResponseCodeEnum.METHOD_ERROR.code)
except BusinessException as e:
return JsonResponse(CommonResult.error(e.code, e.message), json_dumps_params={'ensure_ascii': False})
四、前端代码
<template>
<div class="app-container">
<el-card shadow="always">
<el-row :gutter="24">
<el-input v-model="sqlParam.sql" type="textarea" :rows="10" placeholder="请输入SQL查询" />
<el-button type="primary" @click="executeSQL">执行</el-button>
</el-row>
</el-card>
<el-card shadow="always">
<el-table v-if="data.length > 0" :data="data" style="width: 100%">
<el-table-column v-for="key in Object.keys(data[0])" :key="key" :prop="key" :label="key" />
</el-table>
</el-card>
</div>
</template>
<script>
import {
executeSQLView
} from '@/api/dataBase/database-request'
export default {
data() {
return {
sqlParam: { // 输入框给出默认值
sql: ''
},
data: [],
error: ''
}
},
methods: {
executeSQL() {
this.dataLoading = true
console.log('请求参数:' + this.sqlParam)
executeSQLView(this.sqlParam).then((res) => {
// alert(JSON.stringify(res.data.data))
this.data = res.data.data
setTimeout(() => { // 超过指定超时时间 关闭查询的转圈的loading加载动画
this.listLoading = false
}, 1.5 * 1000)
})
}
}
}
</script>
<style>
</style>
五、运行效果
