refactor: Add logs to the system operation method

This commit is contained in:
wangjiahao
2025-12-26 17:14:34 +08:00
parent 5c88ff8054
commit 1df98cec0f
21 changed files with 971 additions and 27 deletions
+2 -2
View File
@@ -21,8 +21,8 @@ from apps.system.schemas.permission import SqlbotPermission, require_permissions
from common.core.deps import CurrentAssistant, SessionDep, CurrentUser, Trans
from common.utils.command_utils import parse_quick_command
from common.utils.data_format import DataFormat
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
router = APIRouter(tags=["Data Q&A"], prefix="/chat")
+2 -2
View File
@@ -3,8 +3,8 @@ from fastapi import APIRouter, File, UploadFile, HTTPException
from apps.dashboard.crud.dashboard_service import list_resource, load_resource, \
create_resource, create_canvas, validate_name, delete_resource, update_resource, update_canvas
from apps.dashboard.models.dashboard_model import CreateDashboard, BaseDashboard, QueryDashboard, DashboardResponse
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
from common.core.deps import SessionDep, CurrentUser
router = APIRouter(tags=["dashboard"], prefix="/dashboard")
@@ -19,8 +19,8 @@ from common.core.config import settings
from common.core.deps import SessionDep, CurrentUser, Trans
from common.utils.data_format import DataFormat
from common.utils.excel import get_excel_column_count
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
router = APIRouter(tags=["SQL Examples"], prefix="/system/data-training")
+2 -2
View File
@@ -28,8 +28,8 @@ from ..crud.field import get_fields_by_table_id
from ..crud.table import get_tables_by_ds_id
from ..models.datasource import CoreDatasource, CreateDatasource, TableObj, CoreTable, CoreField, FieldObj, \
TableSchemaResponse, ColumnSchemaResponse, PreviewResponse
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
router = APIRouter(tags=["Datasource"], prefix="/datasource")
path = settings.EXCEL_PATH
@@ -5,8 +5,8 @@ from apps.datasource.crud.recommended_problem import get_datasource_recommended,
save_recommended_problem, get_datasource_recommended_base
from apps.datasource.models.datasource import RecommendedProblemBase
from common.core.deps import SessionDep, CurrentUser
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
router = APIRouter(tags=["recommended_problem"], prefix="/recommended_problem")
+2 -2
View File
@@ -16,8 +16,8 @@ from common.utils.time import get_timestamp
from common.utils.utils import SQLBotLogUtil, prepare_model_arg
router = APIRouter(tags=["system_model"], prefix="/system/aimodel")
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
@router.post("/status", include_in_schema=False)
@require_permissions(permission=SqlbotPermission(role=['admin']))
+2 -2
View File
@@ -9,8 +9,8 @@ from common.utils.time import get_timestamp
import secrets
router = APIRouter(tags=["system_apikey"], prefix="/system/apikey", include_in_schema=False)
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
@router.get("")
async def grid(session: SessionDep, current_user: CurrentUser) -> list[ApikeyGridItem]:
+2 -2
View File
@@ -21,8 +21,8 @@ from common.core.sqlbot_cache import clear_cache
from common.utils.utils import get_origin_from_referer, origin_match_domain
router = APIRouter(tags=["system_assistant"], prefix="/system/assistant")
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
@router.get("/info/{id}", include_in_schema=False)
+2 -2
View File
@@ -12,8 +12,8 @@ from common.core.config import settings
from common.core.schemas import Token
from sqlbot_xpack.authentication.manage import logout as xpack_logout
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import system_log, LogConfig
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import system_log, LogConfig
router = APIRouter(tags=["login"], prefix="/login")
+2 -2
View File
@@ -6,8 +6,8 @@ from apps.system.schemas.permission import SqlbotPermission, require_permissions
from common.core.deps import SessionDep
router = APIRouter(tags=["system/parameter"], prefix="/system/parameter", include_in_schema=False)
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
@router.get("/login")
async def get_login_args(session: SessionDep) -> list[SysArgModel]:
+2 -2
View File
@@ -9,8 +9,8 @@ from apps.system.models.user import UserModel
from apps.system.schemas.auth import CacheName, CacheNamespace
from apps.system.schemas.permission import SqlbotPermission, require_permissions
from apps.system.schemas.system_schema import PwdEditor, UserCreator, UserEditor, UserGrid, UserInfoDTO, UserLanguage, UserStatus, UserWs
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
from common.core.deps import CurrentUser, SessionDep, Trans
from common.core.pagination import Paginator
from common.core.schemas import PaginatedResponse, PaginationParams
+2 -2
View File
@@ -8,8 +8,8 @@ from apps.system.models.system_model import UserWsModel, WorkspaceBase, Workspac
from apps.system.models.user import UserModel
from apps.system.schemas.permission import SqlbotPermission, require_permissions
from apps.system.schemas.system_schema import UserWsBase, UserWsDTO, UserWsEditor, UserWsOption, WorkspaceUser
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import system_log, LogConfig
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import system_log, LogConfig
from common.core.deps import CurrentUser, SessionDep, Trans
from common.core.pagination import Paginator
from common.core.schemas import PaginatedResponse, PaginationParams
+2 -2
View File
@@ -19,8 +19,8 @@ from common.core.config import settings
from common.core.deps import SessionDep, CurrentUser, Trans
from common.utils.data_format import DataFormat
from common.utils.excel import get_excel_column_count
from sqlbot_xpack.audit.models.log_model import OperationType, OperationModules
from sqlbot_xpack.audit.schemas.logger_decorator import LogConfig, system_log
from common.audit.models.log_model import OperationType, OperationModules
from common.audit.schemas.logger_decorator import LogConfig, system_log
router = APIRouter(tags=["Terminology"], prefix="/system/terminology")
View File
+90
View File
@@ -0,0 +1,90 @@
from sqlmodel import Field, SQLModel
from datetime import datetime
from typing import Optional
from enum import Enum
from pydantic import BaseModel
class OperationModules(str, Enum):
CHAT = "chat" # 问数
DATASOURCE = "datasource" # 数据源
DASHBOARD = "dashboard" # 仪表板
MEMBER = "member" # 成员
PERMISSION = "permission" # 权限
RULES = "rules" # q组
TERMINOLOGY = "terminology" # 术语
DATA_TRAINING = "data_training" # SQL 示例库
PROMPT_WORDS = "prompt_words" # 自定义提示词
USER = "user" # 用户
WORKSPACE = "workspace" # 工作空间
AI_MODEL = "ai_model" # AI 模型
APPLICATION = "application" # 嵌入式管理 应用
THEME = "theme" # 外观配置
PARAMS_SETTING = "params_setting" # 参数配置
API_KEY = "api_key" # api key
LOG_SETTING = "log_setting" # api key
class OperationStatus(str, Enum):
SUCCESS = "success"
FAILED = "failed"
class OperationType(str, Enum):
CREATE = "create"
DELETE = "delete"
UPDATE = "update"
EDIT = "edit"
LOGIN = "login"
VIEW = "view"
EXPORT = "export"
IMPORT = "import"
ADD = "add"
CREATE_OR_UPDATE = "create_or_update"
class SystemLog(SQLModel, table=True):
__tablename__ = "sys_logs"
id: Optional[int] = Field(default=None, primary_key=True)
operation_type: str = Field(default=None)
operation_detail: str = Field(default=None)
user_id: Optional[int] = Field(default=None)
operation_status: str = Field(default=None)
ip_address: Optional[str] = Field(default=None)
user_agent: Optional[str] = Field(default=None)
execution_time: int = Field(default=0, description="执行时间(毫秒)")
error_message: Optional[str] = Field(default=None)
create_time: datetime = Field(default_factory=datetime.now)
module: Optional[str] = Field(default=None)
oid: Optional[int] = Field(default=None)
resource_id: Optional[str] = Field(default=None)
request_method: Optional[str] = Field(default=None)
request_path: Optional[str] = Field(default=None)
remark: Optional[str] = Field(default=None)
user_name: Optional[str] = Field(default=None)
resource_name: Optional[str] = Field(default=None)
class SystemLogInfo(BaseModel):
id: str = Field(default=None)
operation_type_name: str = Field(default=None)
operation_detail_info: str = Field(default=None)
user_name: str = Field(default=None)
resource_name: str = Field(default=None)
operation_status: str = Field(default=None)
ip_address: Optional[str] = Field(default=None)
create_time: datetime = Field(default_factory=datetime.now)
oid_list: str = Field(default=None)
remark: str = Field(default=None)
class SystemLogInfoResult(BaseModel):
id: str = Field(default=None)
operation_type_name: str = Field(default=None)
operation_detail_info: str = Field(default=None)
user_name: str = Field(default=None)
resource_name: str = Field(default=None)
operation_status: str = Field(default=None)
operation_status_name: str = Field(default=None)
ip_address: Optional[str] = Field(default=None)
create_time: datetime = Field(default_factory=datetime.now)
oid_name: str = Field(default=None)
oid: str = Field(default=None)
error_message: str = Field(default=None)
remark: str = Field(default=None)
+144
View File
@@ -0,0 +1,144 @@
from sqlalchemy import select, func
from sqlalchemy.sql import Select
from sqlalchemy import String, union_all
from apps.chat.models.chat_model import Chat
from apps.dashboard.models.dashboard_model import CoreDashboard
from apps.data_training.models.data_training_model import DataTraining
from apps.datasource.models.datasource import CoreDatasource
from apps.system.models.system_model import WorkspaceModel, AiModelDetail, ApiKeyModel
from apps.system.models.user import UserModel
from apps.terminology.models.terminology_model import Terminology
from apps.system.models.system_model import AssistantModel
from sqlbot_xpack.permissions.models.ds_rules import DsRules
from sqlbot_xpack.custom_prompt.models.custom_prompt_model import CustomPrompt
from sqlbot_xpack.permissions.models.ds_permission import DsPermission
from sqlalchemy import literal_column
def build_resource_union_query() -> Select:
"""
构建资源名称的union查询
返回包含id, name, module的查询
"""
# 创建各个子查询,每个查询都包含module字段
# ai_model 表查询
ai_model_query = select(
func.cast(AiModelDetail.id, String).label("id"),
AiModelDetail.name.label("name"),
literal_column("'ai_model'").label("module")
).select_from(AiModelDetail)
# chat 表查询(使用brief作为name)
chat_query = select(
func.cast(Chat.id, String).label("id"),
Chat.brief.label("name"),
literal_column("'chat'").label("module")
).select_from(Chat)
# dashboard 表查询
dashboard_query = select(
func.cast(CoreDashboard.id, String).label("id"),
CoreDashboard.name.label("name"),
literal_column("'dashboard'").label("module")
).select_from(CoreDashboard)
# datasource 表查询
datasource_query = select(
func.cast(CoreDatasource.id, String).label("id"),
CoreDatasource.name.label("name"),
literal_column("'datasource'").label("module")
).select_from(CoreDatasource)
# custom_prompt 表查询
custom_prompt_query = select(
func.cast(CustomPrompt.id, String).label("id"),
CustomPrompt.name.label("name"),
literal_column("'prompt_words'").label("module")
).select_from(CustomPrompt)
# data_training 表查询(使用question作为name)
data_training_query = select(
func.cast(DataTraining.id, String).label("id"),
DataTraining.question.label("name"),
literal_column("'data_training'").label("module")
).select_from(DataTraining)
# ds_permission 表查询
ds_permission_query = select(
func.cast(DsPermission.id, String).label("id"),
DsPermission.name.label("name"),
literal_column("'permission'").label("module")
).select_from(DsPermission)
# ds_rules 表查询
ds_rules_query = select(
func.cast(DsRules.id, String).label("id"),
DsRules.name.label("name"),
literal_column("'rules'").label("module")
).select_from(DsRules)
# sys_user 表查询
user_query = select(
func.cast(UserModel.id, String).label("id"),
UserModel.name.label("name"),
literal_column("'user'").label("module")
).select_from(UserModel)
# sys_user 表查询
member_query = select(
func.cast(UserModel.id, String).label("id"),
UserModel.name.label("name"),
literal_column("'member'").label("module")
).select_from(UserModel)
# sys_workspace 表查询
sys_workspace_query = select(
func.cast(WorkspaceModel.id, String).label("id"),
WorkspaceModel.name.label("name"),
literal_column("'workspace'").label("module")
).select_from(WorkspaceModel)
# terminology 表查询(使用word作为name)
terminology_query = select(
func.cast(Terminology.id, String).label("id"),
Terminology.word.label("name"),
literal_column("'terminology'").label("module")
).select_from(Terminology)
# sys_assistant 表查询
sys_assistant_query = select(
func.cast(AssistantModel.id, String).label("id"),
AssistantModel.name.label("name"),
literal_column("'application'").label("module")
).select_from(AssistantModel)
# sys_apikey 表查询
sys_apikey_query = select(
func.cast(ApiKeyModel.id, String).label("id"),
ApiKeyModel.access_key.label("name"),
literal_column("'api_key'").label("module")
).select_from(ApiKeyModel)
# 使用 union_all() 方法连接所有查询
union_query = union_all(
ai_model_query,
chat_query,
dashboard_query,
datasource_query,
custom_prompt_query,
data_training_query,
ds_permission_query,
ds_rules_query,
user_query,
member_query,
sys_workspace_query,
terminology_query,
sys_assistant_query,
sys_apikey_query
)
# 返回查询,包含所有字段
return select(union_query.c.id, union_query.c.name, union_query.c.module)
@@ -0,0 +1,673 @@
import time
import functools
import json
import inspect
from typing import Callable, Any, Optional, Dict, Union
from fastapi import Request, HTTPException
from datetime import datetime
from pydantic import BaseModel
from sqlmodel import Session, select
import traceback
from sqlbot_xpack.audit.curd.audit import build_resource_union_query
from common.audit.models.log_model import OperationType, OperationStatus, SystemLog
from common.audit.schemas.request_context import RequestContext
from apps.system.crud.user import get_user_by_account
from apps.system.schemas.system_schema import UserInfoDTO, BaseUserDTO
from sqlalchemy import and_, select
from common.core.db import engine
def get_resource_name_by_id_and_module(session, resource_id: Any, module: str) -> Optional[str]:
resource_union_query = build_resource_union_query()
resource_alias = resource_union_query.alias("resource")
if isinstance(resource_id, list):
# 处理列表情况
if not resource_id:
return None
# 构建查询,使用 IN 条件
query = select(
resource_alias.c.name
).where(
and_(
resource_alias.c.id.in_([str(id_) for id_ in resource_id]),
resource_alias.c.module == module
)
)
# 执行查询并获取所有结果
results = session.execute(query).scalars().all()
# 用逗号连接所有名称
return ",".join(results) if results else None
else:
# 处理单个字符串情况
query = select(
resource_alias.c.name
).where(
and_(
resource_alias.c.id == str(resource_id),
resource_alias.c.module == module
)
)
return session.execute(query).scalar()
class LogConfig(BaseModel):
operation_type: OperationType
operation_detail: str = None
module: Optional[str] = None
# Extract the expression of resource ID from the parameters
resource_id_expr: Optional[str] = None
# Extract the expression for resource ID from the returned result
result_id_expr: Optional[str] = None
# Extract the expression for resource name or other info from the returned result
remark_expr: Optional[str] = None
# Is it only recorded upon success
save_on_success_only: bool = False
# Whether to ignore errors (i.e. record them as successful even when they occur)
ignore_errors: bool = False
# Whether to extract request parameters
extract_params: bool = True
# Delay recording (if the resource ID needs to be extracted from the result, it can be set to True)
delay_logging: bool = False
class SystemLogger:
@staticmethod
async def create_log(
session: Session,
operation_type: OperationType,
operation_detail: str,
user: Optional[UserInfoDTO] = None,
status: OperationStatus = OperationStatus.SUCCESS,
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
execution_time: int = 0,
error_message: Optional[str] = None,
module: Optional[str] = None,
resource_id: Any = None,
request_method: Optional[str] = None,
request_path: Optional[str] = None,
remark: Optional[str] = None
):
try:
log = SystemLog(
operation_type=operation_type,
operation_detail=operation_detail,
user_id=user.id if user else None,
user_name=user.username if user else None,
operation_status=status,
ip_address=ip_address,
user_agent=user_agent,
execution_time=execution_time,
error_message=error_message,
module=module,
resource_id=resource_id,
request_method=request_method,
request_path=request_path,
created_at=datetime.now(),
remark=remark
)
session.add(log)
session.commit()
return log
except Exception as e:
session.rollback()
print(f"[SystemLogger] Log -14 : {str(traceback.format_exc())}")
print(f"Failed to create system log: {e}")
return None
@staticmethod
def get_client_info(request: Request) -> Dict[str, Optional[str]]:
"""Obtain client information"""
ip_address = None
user_agent = None
if request:
# Obtain IP address
if request.client:
ip_address = request.client.host
# Attempt to obtain the real IP from X-Forwarded-For
if "x-forwarded-for" in request.headers:
ip_address = request.headers["x-forwarded-for"].split(",")[0].strip()
# Get User Agent
user_agent = request.headers.get("user-agent")
return {
"ip_address": ip_address,
"user_agent": user_agent
}
@staticmethod
def extract_value_from_object(expression: str, obj: Any):
"""
Extract values from objects based on expressions
support:
-Object attribute: 'user. id'
-Dictionary key: 'data ['id'] '
-List index: 'items [0]. id'
"""
if not expression or obj is None:
return None
if expression == 'result_self':
return obj
try:
# Handling point separated attribute access
parts = expression.split('.')
current = obj
for part in parts:
if not current:
return None
# Handle dictionary key access, such as data ['id ']
if '[' in part and ']' in part:
import re
# Extract key names, such as data ['id '] ->key='id'
match = re.search(r"\[['\"]?([^\]'\"\]]+)['\"]?\]", part)
if match:
key = match.group(1)
# Get Object Part
obj_part = part.split('[')[0]
if hasattr(current, obj_part):
current = getattr(current, obj_part)
elif isinstance(current, dict) and obj_part in current:
current = current[obj_part]
else:
return None
# Get key value
if isinstance(current, dict) and key in current:
current = current[key]
elif hasattr(current, key):
current = getattr(current, key)
elif isinstance(current, list) and key.isdigit():
index = int(key)
if 0 <= index < len(current):
current = current[index]
else:
return None
else:
return None
else:
return None
# Process list indexes, such as items.0.id
elif part.isdigit() and isinstance(current, (list, tuple)):
index = int(part)
if 0 <= index < len(current):
current = current[index]
else:
return None
# Normal attribute access
else:
if hasattr(current, part):
current = getattr(current, part)
elif isinstance(current, dict) and part in current:
current = current[part]
else:
return None
return current if current is not None else None
except Exception:
return None
@staticmethod
def extract_resource_id(
expression: Optional[str],
source: Any,
source_type: str = "args" # args, kwargs, result
):
"""Extract resource IDs from different sources"""
if not expression:
return None
try:
if source_type == "result":
# Extract directly from the result object
return SystemLogger.extract_value_from_object(expression, source)
elif source_type == "args":
# Extract from function parameters
if isinstance(source, tuple) and len(source) > 0:
# The first element is the function itself
func_args = source[0] if isinstance(source[0], tuple) else source
# Processing args [index] expression
if expression.startswith("args["):
import re
pattern = r"args\[(\d+)\]"
match = re.match(pattern, expression)
if match:
index = int(match.group(1))
if index < len(func_args):
value = func_args[index]
return value if value is not None else None
# Process attribute expressions
return SystemLogger.extract_value_from_object(expression, func_args)
elif source_type == "kwargs":
# Extract from keyword parameters
if isinstance(source, dict):
# Simple parameter name
if expression in source:
value = source[expression]
return value if value is not None else None
# complex expression
return SystemLogger.extract_value_from_object(expression, source)
return None
except Exception:
return None
@staticmethod
def extract_from_function_params(
expression: Optional[str],
func_args: tuple,
func_kwargs: dict
):
"""Extract values from function parameters"""
if not expression:
return None
# Attempt to extract from location parameters
result = SystemLogger.extract_resource_id(expression, func_args, "args")
if result:
return result
# Attempt to extract from keyword parameters
result = SystemLogger.extract_resource_id(expression, func_kwargs, "kwargs")
if result:
return result
# Attempt to encapsulate parameters as objects for extraction
try:
if func_args:
# Create a dictionary containing all parameters
params_dict = {}
# Add location parameters
for i, arg in enumerate(func_args):
params_dict[f"arg_{i}"] = arg
# Add keyword parameters
params_dict.update(func_kwargs)
# Attempt to extract from the dictionary
return SystemLogger.extract_resource_id(expression, params_dict, "kwargs")
except:
pass
return None
@staticmethod
def get_current_user(request: Optional[Request]):
"""Retrieve current user information from the request"""
if not request:
return None
try:
current_user = getattr(request.state, 'current_user', None)
if current_user:
return current_user
except:
pass
return None
@staticmethod
def extract_request_params(request: Optional[Request]):
"""Extract request parameters"""
if not request:
return None
try:
params = {}
# query parameters
if request.query_params:
params["query"] = dict(request.query_params)
# path parameter
if request.path_params:
params["path"] = dict(request.path_params)
# Head information (sensitive information not recorded)
headers = {}
for key, value in request.headers.items():
if key.lower() not in ["authorization", "cookie", "set-cookie"]:
headers[key] = value
if headers:
params["headers"] = headers
# Request Body - Only records Content Type and Size
content_type = request.headers.get("content-type", "")
content_length = request.headers.get("content-length")
params["body_info"] = {
"content_type": content_type,
"content_length": content_length
}
return json.dumps(params, ensure_ascii=False, default=str)
except Exception:
return None
@classmethod
async def create_log_record(
cls,
config: LogConfig,
status: OperationStatus,
execution_time: int,
error_message: Optional[str] = None,
resource_id: Any = None,
resource_name: Optional[str] = None,
request: Optional[Request] = None,
remark: Optional[str] = None,
oid: int = -1,
opt_type_ref : OperationType = None,
) -> Optional[SystemLog]:
"""Create log records"""
try:
print(f"[SystemLogger] Log -9")
# Obtain user information
user_info = cls.get_current_user(request)
user_id = user_info.id if user_info else -1
user_name = user_info.name if user_info else '-1'
if config.operation_type == OperationType.LOGIN:
user_id = resource_id
user_name = resource_name
# Obtain client information
client_info = cls.get_client_info(request)
# Get request parameters
request_params = None
if config.extract_params:
request_params = cls.extract_request_params(request)
print(f"[SystemLogger-Info] :user_info {str(user_info)},oid {str(oid)}")
# Create log object
log = SystemLog(
operation_type=opt_type_ref if opt_type_ref else config.operation_type,
operation_detail=config.operation_detail,
user_id=user_id,
user_name=user_name,
oid=user_info.oid if user_info else oid,
operation_status=status,
ip_address=client_info.get("ip_address"),
user_agent=client_info.get("user_agent"),
execution_time=execution_time,
error_message=error_message,
module=config.module,
resource_id=str(resource_id),
resource_name=str(resource_name),
request_method=request.method if request else None,
request_path=request.url.path if request else None,
request_params=request_params,
create_time=datetime.now(),
remark=remark
)
with Session(engine) as session:
print(f"[SystemLogger] Log -10")
session.add(log)
session.commit()
session.refresh(log)
return log
except Exception as e:
print(f"[SystemLogger] Failed to create log: {str(traceback.format_exc())}")
return None
def system_log(config: Union[LogConfig, Dict]):
"""
System log annotation decorator, supports extracting resource IDs from returned results
Usage example:
@system_log({
"operation_type": OperationType.CREATE,
Operation_detail ":" Create User ",
"module": "user",
'result_id_expr ':' id '# Extract the id field from the returned result
})
"""
# If a dictionary is passed in, convert it to a LogConfig object
if isinstance(config, dict):
config = LogConfig(**config)
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
status = OperationStatus.SUCCESS
error_message = None
request = None
resource_id = None
resource_name = None
remark = None
oid = -1
opt_type_ref = None
try:
# Get current request
request = RequestContext.get_request()
print(f"[SystemLogger] Log -1 : {str(request)}")
# Step 1: Attempt to extract the resource ID from the parameters
if config.resource_id_expr:
resource_id = SystemLogger.extract_from_function_params(
config.resource_id_expr,
args,
kwargs
)
print(f"[SystemLogger] Log -2 : {str(resource_id)}")
if config.remark_expr:
remark = SystemLogger.extract_from_function_params(
config.remark_expr,
args,
kwargs
)
if config.operation_type == OperationType.LOGIN:
input_account_dec = SystemLogger.extract_from_function_params(
"form_data.username",
args,
kwargs
)
from common.utils.crypto import sqlbot_decrypt
input_account = await sqlbot_decrypt(input_account_dec)
with Session(engine) as session:
userInfo = get_user_by_account(session=session, account=input_account)
if userInfo is not None:
resource_id = userInfo.id
resource_name = userInfo.name
oid = userInfo.oid
else:
resource_id = -1
oid = -1
resource_name = '-' + input_account
print(f"[SystemLogger] Log -3 : {str(resource_id)}")
if config.operation_type == OperationType.DELETE:
with Session(engine) as session:
resource_name = get_resource_name_by_id_and_module(session, resource_id, config.module)
if config.operation_type == OperationType.CREATE_OR_UPDATE:
opt_type_ref = OperationType.UPDATE if resource_id is not None else OperationType.CREATE
else:
opt_type_ref = config.operation_type
print(f"[SystemLogger] Log -4 : {str(opt_type_ref)}")
# Execute the original function
result = await func(*args, **kwargs)
print(f"[SystemLogger] Log -5 : {str(result)}")
# Step 2: If the resource ID is configured to be extracted from the results and has not been extracted before
if config.result_id_expr and not resource_id and result:
resource_id = SystemLogger.extract_resource_id(
config.result_id_expr,
result,
"result"
)
if config.operation_type != OperationType.DELETE:
with Session(engine) as session:
resource_name = get_resource_name_by_id_and_module(session, resource_id, config.module)
print(f"[SystemLogger] Log -6 : {str(resource_name)}")
return result
except Exception as e:
print(f"[SystemLogger] Log -6 : {str(traceback.format_exc())}")
status = OperationStatus.FAILED
error_message = str(e)
# If it is an HTTPException, retrieve the status code
if isinstance(e, HTTPException):
error_message = f"HTTP {e.status_code}: {e.detail}"
# If configured to ignore errors, mark as successful
if config.ignore_errors:
status = OperationStatus.SUCCESS
raise e
finally:
# If configured to only record on success and the current status is failure, skip
if config.save_on_success_only and status == OperationStatus.FAILED:
return
# Calculate execution time
execution_time = int((time.time() - start_time) * 1000)
print(f"[SystemLogger] Log -8 ")
# Asynchronous creation of log records
try:
await SystemLogger.create_log_record(
config=config,
status=status,
execution_time=execution_time,
error_message=error_message,
resource_id=resource_id,
resource_name=resource_name,
remark=remark,
request=request,
oid=oid,
opt_type_ref=opt_type_ref
)
except Exception as log_error:
print(f"[SystemLogger] Log -12 : {str(traceback.format_exc())}")
print(f"[SystemLogger] Log creation failed: {log_error}")
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
start_time = time.time()
status = OperationStatus.SUCCESS
error_message = None
request = None
resource_id = None
resource_name = None
result = None
try:
# Get current request
request = RequestContext.get_request()
# Extract resource ID from parameters
if config.resource_id_expr:
resource_id = SystemLogger.extract_from_function_params(
config.resource_id_expr,
args,
kwargs
)
# Obtain client information
if config.operation_type == OperationType.DELETE:
with Session(engine) as session:
resource_name = get_resource_name_by_id_and_module(session, resource_id, config.module)
# Execute the original function
result = func(*args, **kwargs)
# Extract resource ID from the results
if config.result_id_expr and not resource_id and result:
resource_id = SystemLogger.extract_resource_id(
config.result_id_expr,
result,
"result"
)
if config.operation_type != OperationType.DELETE:
with Session(engine) as session:
resource_name = get_resource_name_by_id_and_module(session, resource_id, config.module)
return result
except Exception as e:
status = OperationStatus.FAILED
error_message = str(e)
if isinstance(e, HTTPException):
error_message = f"HTTP {e.status_code}: {e.detail}"
if config.ignore_errors:
status = OperationStatus.SUCCESS
raise e
finally:
if config.save_on_success_only and status == OperationStatus.FAILED:
return
execution_time = int((time.time() - start_time) * 1000)
# In the synchronous version, we still create logs asynchronously
import asyncio
try:
loop = asyncio.get_event_loop()
if loop.is_running():
loop.create_task(
SystemLogger.create_log_record(
config=config,
status=status,
execution_time=execution_time,
error_message=error_message,
resource_id=resource_id,
resource_name=resource_name,
request=request
)
)
else:
asyncio.run(
SystemLogger.create_log_record(
config=config,
status=status,
execution_time=execution_time,
error_message=error_message,
resource_id=resource_id,
request=request
)
)
except Exception as log_error:
print(f"[SystemLogger] Log -13 : {str(traceback.format_exc())}")
print(f"[SystemLogger] Log creation failed: {log_error}")
# Return appropriate wrapper based on function type
if inspect.iscoroutinefunction(func):
return async_wrapper
else:
return sync_wrapper
return decorator
@@ -0,0 +1,37 @@
from contextvars import ContextVar
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
class RequestContext:
_current_request: ContextVar[Request] = ContextVar("_current_request")
@classmethod
def set_request(cls, request: Request):
return cls._current_request.set(request)
@classmethod
def get_request(cls) -> Request:
try:
return cls._current_request.get()
except LookupError:
raise RuntimeError(
"No request context found. "
"Make sure RequestContextMiddleware is installed."
)
@classmethod
def reset(cls, token):
cls._current_request.reset(token)
class RequestContextMiddlewareCommon(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 设置请求上下文
token = RequestContext.set_request(request)
try:
response = await call_next(request)
return response
finally:
RequestContext.reset(token)
+1 -1
View File
@@ -21,7 +21,7 @@ from apps.system.crud.aimodel_manage import async_model_info
from apps.system.crud.assistant import init_dynamic_cors
from apps.system.middleware.auth import TokenMiddleware
from apps.system.schemas.permission import RequestContextMiddleware
from sqlbot_xpack.audit.schemas.request_context import RequestContextMiddlewareCommon
from common.audit.schemas.request_context import RequestContextMiddlewareCommon
from common.core.config import settings
from common.core.response_middleware import ResponseMiddleware, exception_handler
from common.core.sqlbot_cache import init_sqlbot_cache