添加sql执行历史记录功能，包装join curd功能

app/crud/__init__.py

@@ -81,14 +81,14 @@ def decorator(func):
         async def wrapper(cls, *args, **kwargs):
             try:
                 session: AsyncSession = kwargs.pop("session", None)
-                begin = kwargs.get("begin")
+                nb = kwargs.get("not_begin")
                 if session is not None:
-                    if transaction and begin:
+                    if transaction and not nb:
                         async with session.begin():
                             return await func(cls, *args, session=session, **kwargs)
                     return await func(cls, *args[1:], session=session, **kwargs)
                 async with async_session() as ss:
-                    if transaction and begin:
+                    if transaction and not nb:
                         async with ss.begin():
                             return await func(cls, *args, session=ss, **kwargs)
                     return await func(cls, *args, session=ss, **kwargs)
@@ -159,7 +159,7 @@ def llike(s: str):
     #         raise Exception(f"获取数据失败")
 
     @staticmethod
-    async def pagination(page: int, size: int, session, sql: str, scalars=True):
+    async def pagination(page: int, size: int, session, sql: str, scalars=True, **kwargs):
         """
         分页查询
         :param scalars:
@@ -170,12 +170,13 @@ async def pagination(page: int, size: int, session, sql: str, scalars=True):
         :return:
         """
         data = await session.execute(sql)
+        print(sql)
         total = data.raw.rowcount
         if total == 0:
             return [], 0
         sql = sql.offset((page - 1) * size).limit(size)
         data = await session.execute(sql)
-        if scalars:
+        if scalars and kwargs.get("_join") is None:
             return data.scalars().all(), total
         return data.all(), total
 
@@ -227,7 +228,7 @@ def delete_model(dist, update_user):
     @classmethod
     @RedisHelper.cache("dao")
     @connect
-    async def list_record_with_pagination(cls, page, size, /, *, session=None, **kwargs):
+    async def list_with_pagination(cls, page, size, /, *, session=None, **kwargs):
         """
         通过分页获取数据
         :param session:
@@ -236,7 +237,7 @@ async def list_record_with_pagination(cls, page, size, /, *, session=None, **kwa
         :param kwargs:
         :return:
         """
-        return await cls.pagination(page, size, session, cls.query_wrapper(**kwargs))
+        return await cls.pagination(page, size, session, cls.query_wrapper(**kwargs), **kwargs)
 
     @classmethod
     def where(cls, param: Any, sentence, condition: list):
@@ -271,6 +272,8 @@ def query_wrapper(cls, condition=None, **kwargs):
         if getattr(cls.__model__, "deleted_at", None):
             conditions.append(getattr(cls.__model__, "deleted_at") == 0)
         _sort = kwargs.pop("_sort", None)
+        _select = kwargs.pop("_select", list())
+        _join = kwargs.pop("_join", None)
         # 遍历参数，当参数不为None的时候传递
         for k, v in kwargs.items():
             # 判断是否是like的情况
@@ -280,11 +283,15 @@ def query_wrapper(cls, condition=None, **kwargs):
             # 如果是like模式，则使用Model.字段.like 否则用 Model.字段 等于
             cls.where(v, getattr(cls.__model__, k).like(v) if like else getattr(cls.__model__, k) == v,
                       conditions)
-        sql = select(cls.__model__).where(*conditions)
+        sql = select(cls.__model__, *_select)
+        if isinstance(_join, Iterable):
+            for j in _join:
+                sql = sql.outerjoin(*j)
+        where = sql.where(*conditions)
         if _sort and isinstance(_sort, Iterable):
             for d in _sort:
-                sql = getattr(sql, "order_by")(d)
-        return sql
+                where = getattr(where, "order_by")(d)
+        return where
 
     @classmethod
     @RedisHelper.cache("dao")
@@ -309,7 +316,7 @@ async def query_record(cls, session: AsyncSession = None, **kwargs):
     @classmethod
     @RedisHelper.up_cache("dao")
     @connect(True)
-    async def insert(cls, *, model: PityBase, session: AsyncSession = None, log=False, begin=True):
+    async def insert(cls, *, model: PityBase, session: AsyncSession = None, log=False, not_begin=False):
         session.add(model)
         await session.flush()
         session.expunge(model)

app/crud/config/DbConfigDao.py

@@ -6,13 +6,14 @@
 from sqlalchemy import select, MetaData, text
 from sqlalchemy.exc import ResourceClosedError
 
-from app.crud import Mapper
+from app.crud import Mapper, ModelWrapper
 from app.crud.config.EnvironmentDao import EnvironmentDao
 from app.handler.encoder import JsonEncoder
 from app.handler.fatcory import PityResponse
 from app.middleware.RedisManager import RedisHelper
-from app.models import async_session, DatabaseHelper, db_helper
+from app.models import async_session, db_helper
 from app.models.database import PityDatabase
+from app.models.sql_log import PitySQLHistory
 from app.schema.database import DatabaseForm
 from app.utils.logger import Log
 
@@ -207,10 +208,12 @@ async def execute(conn, sql):
         async with session() as s:
             async with s.begin():
                 try:
+                    start = time.perf_counter()
                     result = await s.execute(text(sql))
+                    cost = time.perf_counter() - start
                     row_count = result.rowcount
                     ans = result.mappings().all()
-                    return ans
+                    return ans, int(cost * 1000)
                 except ResourceClosedError:
                     # 说明是update或其他语句
                     return [{"rowCount": row_count}]
@@ -233,3 +236,8 @@ async def execute_sql(env: int, name: str, sql: str):
         except Exception as e:
             DbConfigDao.log.error(f"查询数据库配置失败, error: {e}")
             raise Exception(f"执行SQL失败: {e}")
+
+
+@ModelWrapper(PitySQLHistory)
+class PitySQLHistoryDao(Mapper):
+    pass

app/crud/test_case/TestCaseDao.py

@@ -84,7 +84,7 @@ async def _insert(session, case_id: int, user_id: int, form: TestCaseInfo, **fie
                     data = model(**f.dict(), user_id=user_id)
                 else:
                     data = model(**f.dict(), user_id=user_id, case_id=case_id)
-                await md.insert(model=data, session=session, begin=False)
+                await md.insert(model=data, session=session, not_begin=True)
 
     @staticmethod
     async def insert_test_case(session, data: TestCaseInfo, user_id: int) -> TestCase:

app/models/database.py

@@ -1,6 +1,7 @@
 from sqlalchemy import INT, Column, String, UniqueConstraint
 
 from app.models.basic import PityBase
+from app.models.environment import Environment
 
 
 class PityDatabase(PityBase):
@@ -17,6 +18,7 @@ class PityDatabase(PityBase):
     password = Column(String(64), nullable=False)
     database = Column(String(36), nullable=False)
     sql_type = Column(INT, nullable=False, comment="0: mysql 1: postgresql 2: mongo")
+    env_data: Environment
 
     def __init__(self, env, name, host, port, username, password, database, sql_type, user, id=None):
         super().__init__(user, id)

app/models/sql_log.py

@@ -0,0 +1,18 @@
+from sqlalchemy import Column, String, INT
+
+from app.models.basic import PityBase
+from app.models.database import PityDatabase
+
+
+class PitySQLHistory(PityBase):
+    __tablename__ = "pity_sql_history"
+    sql = Column(String(1024), comment="sql语句")
+    elapsed = Column(INT, comment="请求耗时")
+    database_id = Column(INT, comment="操作数据库id")
+    database: PityDatabase
+
+    def __init__(self, sql, elapsed, database_id, user):
+        super().__init__(user)
+        self.sql = sql
+        self.elapsed = elapsed
+        self.database_id = database_id

app/routers/config/gconfig.py

@@ -10,7 +10,7 @@
 
 @router.get("/gconfig/list")
 async def list_gconfig(page: int = 1, size: int = 8, env=None, key: str = "", _=Depends(Permission())):
-    data, total = await GConfigDao.list_record_with_pagination(page, size, env=env, key=key)
+    data, total = await GConfigDao.list_with_pagination(page, size, env=env, key=key)
     return PityResponse.success_with_size(data=data, total=total)
 
 

app/routers/online/sql.py

@@ -1,23 +1,44 @@
 from fastapi import APIRouter, Depends
 
-from app.crud.config.DbConfigDao import DbConfigDao
+from app.crud.config.DbConfigDao import DbConfigDao, PitySQLHistoryDao
 from app.handler.fatcory import PityResponse
+from app.models.database import PityDatabase
+from app.models.environment import Environment
+from app.models.sql_log import PitySQLHistory
 from app.routers import Permission
 from app.schema.online_sql import OnlineSQLForm
 
 router = APIRouter(prefix="/online")
 
 
 @router.post("/sql")
-async def execute_sql(data: OnlineSQLForm, _=Depends(Permission())):
+async def execute_sql(data: OnlineSQLForm, user=Depends(Permission())):
     try:
-        result = await DbConfigDao.online_sql(data.id, data.sql)
+        result, elapsed = await DbConfigDao.online_sql(data.id, data.sql)
         columns, result = PityResponse.parse_sql_result(result)
-        return PityResponse.success(data=dict(result=result, columns=columns))
+        await PitySQLHistoryDao.insert(model=PitySQLHistory(data.sql, elapsed, data.id, user['id']))
+        return PityResponse.success(data=dict(result=result, columns=columns, elapsed=elapsed))
     except Exception as err:
         return PityResponse.failed(err)
 
 
+@router.get("/history/query", summary="获取sql执行历史记录")
+async def query_sql_history(page: int = 1, size: int = 4, _=Depends(Permission())):
+    data, total = await PitySQLHistoryDao.list_with_pagination(page, size,
+                                                               _sort=[PitySQLHistory.created_at.desc()],
+                                                               _select=[PityDatabase, Environment],
+                                                               _join=[(PityDatabase,
+                                                                       PityDatabase.id == PitySQLHistory.database_id),
+                                                                      (Environment, Environment.id == PityDatabase.env)
+                                                                      ])
+    ans = []
+    for history, database, env in data:
+        database.env_info = env
+        history.database = database
+        ans.append(history)
+    return PityResponse.success(dict(data=ans, total=total))
+
+
 @router.get("/tables")
 async def list_tables(_=Depends(Permission())):
     try:

gunicorn.py

@@ -4,7 +4,7 @@
 
 # debug = True
 loglevel = 'debug'
-bind = "0.0.0.0:7777"
+bind = "127.0.0.1:7777"
 pidfile = "logs/gunicorn.pid"
 accesslog = "logs/access.log"
 errorlog = "logs/debug.log"