mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 13:05:18 +00:00
zhangtao
60e6452255
refactor(前端): 统一表单搜索组件样式和布局 style: 优化表格和卡片样式，调整高度和间距 fix(上传): 修复文件上传组件name属性缺失问题 perf(IP定位): 优化IP定位逻辑，增加内网IP判断和重试机制 docs: 更新密码修改成功后的提示信息 chore: 清理无用代码和注释 ``` 这个提交消息包含了多个方面的改进： 1. 新增功能(feat)：为用户表添加了avatar字段 2. 重构(refactor)：统一了前端表单搜索组件的样式和布局 3. 样式优化(style)：调整了表格和卡片的样式 4. 问题修复(fix)：修复了文件上传组件name属性缺失的问题 5. 性能优化(perf)：改进了IP定位服务的逻辑 6. 文档更新(docs)：修改了密码修改成功后的提示信息 7. 杂项(chore)：清理了无用的代码和注释 消息遵循了提交规范，每个部分都简明扼要地描述了变更内容。
101 lines
4.4 KiB
Python
101 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import time
|
|
from typing import Any, Callable, Coroutine
|
|
from fastapi import Request, Response
|
|
from fastapi.routing import APIRoute
|
|
from user_agents import parse
|
|
|
|
from app.api.v1.schemas.system.auth_schema import AuthSchema
|
|
from app.api.v1.schemas.system.operation_log_schema import OperationLogCreateSchema
|
|
from app.api.v1.services.system.operation_log_service import OperationLogService
|
|
from app.core.database import session_connect
|
|
from app.config.setting import settings
|
|
from app.utils.ip_local_util import IpLocalUtil
|
|
from app.core.logger import logger
|
|
|
|
"""
|
|
在 FastAPI 中，route_class 参数用于自定义路由的行为。
|
|
通过设置 route_class，你可以定义一个自定义的路由类，从而在每个路由处理之前或之后执行特定的操作。
|
|
这对于日志记录、权限验证、性能监控等场景非常有用。
|
|
"""
|
|
class OperationLogRoute(APIRoute):
|
|
"""操作日志路由装饰器"""
|
|
|
|
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
|
|
original_route_handler = super().get_route_handler()
|
|
|
|
async def custom_route_handler(request: Request) -> Response:
|
|
start_time = time.time()
|
|
# 请求前的处理
|
|
response: Response = await original_route_handler(request)
|
|
|
|
# 请求后的处理
|
|
if not settings.OPERATION_LOG_RECORD:
|
|
return response
|
|
if request.method not in settings.OPERATION_RECORD_METHOD:
|
|
return response
|
|
route: APIRoute = request.scope.get("route")
|
|
if route.name in settings.IGNORE_OPERATION_FUNCTION:
|
|
return response
|
|
|
|
user_agent = parse(request.headers.get("user-agent"))
|
|
payload = b"{}"
|
|
req_content_type = request.headers.get("Content-Type", "")
|
|
|
|
if 'multipart/form-data' in req_content_type or 'application/x-www-form-urlencoded' in req_content_type:
|
|
payload = ', '.join([f'{k}: {v}' for k, v in (await request.form()).items()])
|
|
else:
|
|
payload = await request.body()
|
|
path_params = request.path_params
|
|
oper_param = {}
|
|
if payload:
|
|
oper_param = payload.decode()
|
|
if path_params:
|
|
oper_param.update(path_params)
|
|
# payload = json.dumps(oper_param, ensure_ascii=False)
|
|
payload = str(oper_param)
|
|
|
|
response_data = response.body if "application/json" in response.headers.get("Content-Type", "") else b"{}"
|
|
process_time = time.time() - start_time
|
|
|
|
# 获取当前用户ID,如果是登录接口则为空
|
|
current_user_id = None
|
|
if "user_id" in request.scope:
|
|
current_user_id = request.scope.get("user_id")
|
|
|
|
request_ip = None
|
|
x_forwarded_for = request.headers.get('X-Forwarded-For')
|
|
if x_forwarded_for:
|
|
# 取第一个 IP 地址，通常为客户端真实 IP
|
|
request_ip = x_forwarded_for.split(',')[0].strip()
|
|
else:
|
|
# 若没有 X-Forwarded-For 头，则使用 request.client.host
|
|
request_ip = request.client.host
|
|
|
|
login_location = await IpLocalUtil.get_ip_location(request_ip)
|
|
|
|
|
|
async with session_connect() as session:
|
|
async with session.begin():
|
|
auth = AuthSchema(db=session)
|
|
|
|
await OperationLogService.create_log_service(data=OperationLogCreateSchema(
|
|
request_path = request.url.path,
|
|
request_method = request.method,
|
|
request_payload = payload,
|
|
request_ip = request_ip,
|
|
login_location=login_location,
|
|
request_os = user_agent.os.family,
|
|
request_browser = user_agent.browser.family,
|
|
response_code = response.status_code,
|
|
response_json = response_data.decode(),
|
|
process_time = process_time,
|
|
description = route.summary,
|
|
creator_id = current_user_id
|
|
), auth = auth)
|
|
|
|
return response
|
|
|
|
return custom_route_handler
|