一、开发环境搭建和配置

pip install mysql-connector-pythonpip 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 loggingfrom 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 statusfrom 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.bodyprint(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 = trueconsole.log('请求参数:' + this.sqlParam)executeSQLView(this.sqlParam).then((res) => {// alert(JSON.stringify(res.data.data))this.data = res.data.datasetTimeout(() => { // 超过指定超时时间  关闭查询的转圈的loading加载动画this.listLoading = false}, 1.5 * 1000)})}}
}
</script><style>
</style>

五、运行效果