DataMiddleware/SourceCode/DataMiddleware/app/stores/log_store.py

from datetime import datetime
from typing import List, Optional, Tuple
from utils.mysql_helper import MySQLHelper
from models.log_data import LogModel

class LogStore:
    def __init__(self):
        self._db_helper = MySQLHelper()

    def insert_log(self, log_data: LogModel):
        sql = "INSERT INTO sys_log (username, operation_type, operation_desc, operation_result, operation_data, operation_ip, operation_module, data_changes, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
        params = (
            log_data.username,
            log_data.operation_type,
            log_data.operation_desc,
            log_data.operation_result,
            log_data.operation_data,
            log_data.operation_ip,
            log_data.operation_module,
            log_data.data_changes,
            log_data.created_at
        )
        with self._db_helper:
            self._db_helper.execute_non_query(sql, params)

    def query_logs_paginated(self, page: int, per_page: int, username: str = None, operation_type: str = None, operation_module: str = None, start_date: datetime = None, end_date: datetime = None) -> Tuple[List[LogModel], int]:
        offset = (page - 1) * per_page
        sql_count = "SELECT COUNT(*) as count FROM sys_log WHERE 1=%s"
        sql = "SELECT id, username, operation_type, operation_desc, operation_result, operation_data, operation_ip, operation_module, data_changes, created_at FROM sys_log WHERE 1=%s"
        params_count = (1,)
        params = (1,)

        if username:
            sql_count += " AND username = %s"
            sql += " AND username = %s"
            params_count += (username,)
            params += (username,)

        if operation_type:
            sql_count += " AND operation_type = %s"
            sql += " AND operation_type = %s"
            params_count += (operation_type,)
            params += (operation_type,)

        if operation_module:
            sql_count += " AND operation_module = %s"
            sql += " AND operation_module = %s"
            params_count += (operation_module,)
            params += (operation_module,)

        if start_date:
            sql_count += " AND created_at >= %s"
            sql += " AND created_at >= %s"
            params_count += (start_date,)
            params += (start_date,)

        if end_date:
            sql_count += " AND created_at <= %s"
            sql += " AND created_at <= %s"
            params_count += (end_date,)
            params += (end_date,)

        sql += " ORDER BY created_at DESC LIMIT %s OFFSET %s"
        params += (per_page, offset)

        with self._db_helper:
            result_count = self._db_helper.fetch_one(sql_count, params_count)
            count = result_count["count"] if result_count else 0
            result = self._db_helper.execute_query(sql, params)
            data = []
            if not result:
                return data, count

            for item in result:
                data.append(
                    LogModel(
                        log_id=item["id"],
                        username=item["username"],
                        operation_type=item["operation_type"],
                        operation_desc=item["operation_desc"],
                        operation_result=item["operation_result"],
                        operation_data=item["operation_data"],
                        operation_ip=item["operation_ip"],
                        operation_module=item["operation_module"],
                        data_changes=item["data_changes"],
                        created_at=item["created_at"]
                    )
                )
            return data, count