from rest_framework.generics import GenericAPIView
from rest_framework.response import Response
from rest_framework import status
from django.db import transaction
from .contants import db_dict
contants.py的内容
(
import cx_Oracle
import pymysql
# 定义一个数据库类型&引擎的字典,
db_dict = {'mysql':pymysql,'Oracle':cx_Oracle}
)
from .models import DataSystem,Rule
class DBconnectView(GenericAPIView):
__DBtype = db_dict
def get(self,request,pk,rule_id):
# 通过传入的id进行对应的数据库链接
self.datas = DataSystem.objects.get(pk=pk)
self.url = self.datas.url
self.username = self.datas.username
self.password = self.datas.password_enc
self.DBname = self.datas.name
self.DBtype = self.__DBtype[self.datas.type]
# 获取check_code规则
self.ruledatas = Rule.objects.get(id=rule_id)
self.check_code = self.ruledatas.check_code
# db = __import__(self.DBtype)
try:
conn = self.DBtype.connect(host=self.url,user=self.username,password=self.password,database=self.DBname)
# 链接成功后创建一个游标
cs_ms = conn.cursor()
except Exception as e:
raise e
else:
# 明显的开启事务
with transaction.atomic():
# 在安全的地方,创建保存点,将来操作数据库失败回滚到此
save_id = transaction.savepoint()
try:
# 获取一个元组
db_ret = cs_ms.execute(self.check_code)
except Exception as e:
transaction.savepoint_rollback(save_id)
raise e
else:
db_set = db_ret.fetchone()
# transaction.savepoint_commit(save_id)
finally:
cs_ms.close()
conn.close()
return Response({'pk':pk,'rule_id':rule_id})