mtsd/backend/app/crud/api_request_log.py

"""API 请求日志 CRUD"""
from datetime import datetime
from typing import List, Optional, Tuple

from sqlalchemy import and_, desc, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_id
from app.models.models import SysApiRequestLog, SysApiRequestLogHis


class ApiRequestLogCRUD:
    @staticmethod
    def get_multi_with_filters(
        db: Session,
        http_method: str = None,
        status_code: int = None,
        keyword: str = None,
        path_prefix: str = None,
        start_date: str = None,
        end_date: str = None,
        skip: int = 0,
        limit: int = 10,
    ) -> Tuple[List[SysApiRequestLog], int]:
        q = db.query(SysApiRequestLog).filter(SysApiRequestLog.is_del == 0)
        if http_method:
            q = q.filter(SysApiRequestLog.http_method == http_method.upper())
        if status_code is not None:
            q = q.filter(SysApiRequestLog.status_code == status_code)
        if path_prefix:
            q = q.filter(SysApiRequestLog.path.like(f"%{path_prefix}%"))
        if keyword:
            kw = f"%{keyword}%"
            q = q.filter(
                or_(
                    SysApiRequestLog.path.like(kw),
                    SysApiRequestLog.username.like(kw),
                    SysApiRequestLog.user_id.like(kw),
                    SysApiRequestLog.request_body.like(kw),
                    SysApiRequestLog.response_body.like(kw),
                )
            )
        if start_date:
            try:
                start_dt = datetime.strptime(start_date, "%Y-%m-%d")
                q = q.filter(SysApiRequestLog.create_at >= start_dt)
            except Exception:
                pass
        if end_date:
            try:
                end_dt = datetime.strptime(end_date, "%Y-%m-%d")
                end_dt = end_dt.replace(hour=23, minute=59, second=59)
                q = q.filter(SysApiRequestLog.create_at <= end_dt)
            except Exception:
                pass

        total = int(
            q.with_entities(func.count(SysApiRequestLog.id)).order_by(None).scalar() or 0
        )
        rows = (
            q.with_entities(
                SysApiRequestLog.id,
                SysApiRequestLog.http_method,
                SysApiRequestLog.path,
                SysApiRequestLog.query_string,
                SysApiRequestLog.status_code,
                SysApiRequestLog.duration_ms,
                SysApiRequestLog.user_id,
                SysApiRequestLog.username,
                SysApiRequestLog.role,
                SysApiRequestLog.ip_address,
                SysApiRequestLog.user_agent,
                func.substr(SysApiRequestLog.request_body, 1, 300).label("request_body"),
                func.substr(SysApiRequestLog.response_body, 1, 300).label("response_body"),
                SysApiRequestLog.create_at,
            )
            .order_by(desc(SysApiRequestLog.create_at))
            .offset(skip)
            .limit(limit)
            .all()
        )
        return rows, total

    @staticmethod
    def get(db: Session, log_id: str) -> Optional[SysApiRequestLog]:
        return db.query(SysApiRequestLog).filter(
            and_(SysApiRequestLog.id == log_id, SysApiRequestLog.is_del == 0)
        ).first()

    @staticmethod
    def delete(db: Session, log_id: str) -> bool:
        log = db.query(SysApiRequestLog).filter(SysApiRequestLog.id == log_id).first()
        if not log:
            return False
        his = SysApiRequestLogHis(
            id=log.id,
            http_method=log.http_method,
            path=log.path,
            query_string=log.query_string,
            request_body=log.request_body,
            response_body=log.response_body,
            status_code=log.status_code,
            duration_ms=log.duration_ms,
            user_id=log.user_id,
            username=log.username,
            role=log.role,
            ip_address=log.ip_address,
            user_agent=log.user_agent,
            create_at=log.create_at,
            create_by=log.create_by,
            create_name=log.create_name,
            update_at=log.update_at,
            update_by=log.update_by,
            his_create_at=datetime.now(),
        )
        db.add(his)
        log.is_del = 1
        db.commit()
        return True


api_request_log_crud = ApiRequestLogCRUD()