2

automated_main/models/api_querrydiff/__init__.py

-# -*- coding: utf-8 -*-

automated_main/models/api_querrydiff/api_querrydiff_task.py

-# -*- coding: utf-8 -*-
-
-from django.db import models
-
-
-class QuerryDiffTask(models.Model):
-    """
-    回放任务
-    """
-    task_name = models.CharField("回放任务名称", max_length=100, blank=False, default="")
-    source_env_id = models.IntegerField("基准环境id", blank=False)
-    target_env_id = models.IntegerField("测试环境id", blank=False)
-    log_time = models.TextField("关联API用例", default="")
-    project_id = models.IntegerField("无效字段去除", blank=False, null=True)
-    simple_num = models.IntegerField("单接口样本量", blank=False)
-
-    urls = models.JSONField("指定url",  blank=True, null=True)
-    opt_num = models.IntegerField("运行次数", default=0, null=True)
-    status = models.IntegerField("状态", default=0, null=True)
-
-    created_name = models.CharField("创建人", max_length=255, blank=True, default="", null=True)
-    created_time = models.DateTimeField("创建时间", auto_now_add=True)
-
-    def __str__(self):
-        return self.task_name

automated_main/models/api_querrydiff/api_querrydiff_task_detail.py

-# -*- coding: utf-8 -*-
-
-from django.db import models
-from automated_main.models.api_querrydiff.api_querrydiff_task import QuerryDiffTask
-
-
-class QuerryDiffTaskItems(models.Model):
-    """
-    回放任务明细
-    """
-    task_id = models.ForeignKey(QuerryDiffTask, on_delete=models.CASCADE)
-    api_url = models.CharField("请求地址", max_length=255, blank=False)
-    param = models.TextField("关联API用例", default=None)
-    result = models.CharField("运行结果",  blank=True, null=True)
-    req_count = models.BigIntegerField("线上运行次数", default=0, null=True)
-    status = models.IntegerField("状态", default=0, null=True)
-    method = models.IntegerField("请求方式", blank=False)
-    created_time = models.DateTimeField("创建时间", auto_now_add=True)
-
-    def __str__(self):
-        return self.task_id

automated_main/models/api_querrydiff/record_opt.py

-# -*- coding: utf-8 -*-
-
-import records
-import traceback
-
-
-class DbRecordOpt:
-    def __init__(self):
-
-        try:
-            self.con = records.Database('mysql+pymysql://root:admin666@172.30.100.253:3306/test_qa')
-        except:
-            print(traceback.print_exc())
-            raise
-
-    def querry_excute(self, order):
-        return self.con.query(order).as_dict()
-
-    def write_excute(self, order):
-        return self.con.query(order)
-
-    def batch(self, order, param):
-        return self.con.bulk_query(order, param)
-
-    def tranc_excute(self, order_list):
-        tr = self.con.transaction()
-        try:
-            for i in order_list:
-                self.con.query(i)
-            tr.commit()
-            return 0
-        except:
-            tr.rollback()
-            return 1