peewee创建连接前的前置操作wireshark抓包实现

from core.mysql.models import ParseMonitoringDetail
from peewee import fn
from loguru import logger
group_by_parser_name: bool = True
group_by_error_class_name: bool = True
group_by_track_source_id: bool = True
query = (ParseMonitoringDetail
         .select(
             ParseMonitoringDetail.parser_name,
             ParseMonitoringDetail.error_class_name,
             ParseMonitoringDetail.track_source_id,
             fn.COUNT("*").alias('task_count')
         )
         .where(ParseMonitoringDetail.success == False)
         .order_by(fn.COUNT("*").alias('task_count').desc()))
if group_by_parser_name:
    query = query.group_by(
        ParseMonitoringDetail.parser_name,
    )
if group_by_error_class_name:
    query = query.group_by(
        ParseMonitoringDetail.error_class_name,
    )
if group_by_track_source_id:
    query = query.group_by(
        ParseMonitoringDetail.track_source_id
    )
logger.debug(f'sql: {str(query)}')
result = query.dicts().execute()
result = list(result)
logger.debug(result)

SELECT
    `t1`.`parser_name`,
    `t1`.`error_class_name`,
    `t1`.`track_source_id`,
    COUNT('*') AS `task_count`
FROM
    `parse_monitoring_detail` AS `t1`
WHERE
    (`t1`.`success` = 0)
GROUP BY
    `t1`.`track_source_id`
ORDER BY
    `task_count` DESC

SET NAMES utf8mb4

SET sql_mode='PIPES_AS_CONCAT'

SELECT
    `t1`.`parser_name`,
    `t1`.`error_class_name`,
    `t1`.`track_source_id`,
    COUNT('*') AS `task_count`
FROM
    `parse_monitoring_detail` AS `t1`
WHERE
    (`t1`.`success` = 0)
GROUP BY
    `t1`.`track_source_id`
ORDER BY
    `task_count` DESC