sqxp/backend/app/crud/api_request_log.py

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

from sqlalchemy import and_, desc, 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 = q.count()
        rows = q.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()