Browse Source

style: 使用ruff格式化annotation模块,优化导入

master
insistence 7 months ago
parent
commit
bdb1a67d44
  1. 64
      ruoyi-fastapi-backend/module_admin/annotation/log_annotation.py
  2. 13
      ruoyi-fastapi-backend/module_admin/annotation/pydantic_annotation.py

64
ruoyi-fastapi-backend/module_admin/annotation/log_annotation.py

@ -1,23 +1,27 @@
from functools import wraps, lru_cache
from fastapi import Request
from fastapi.responses import JSONResponse, ORJSONResponse, UJSONResponse
import inspect import inspect
import os
import json import json
import time import os
from datetime import datetime
import requests import requests
import time
import warnings import warnings
from datetime import datetime
from fastapi import Request
from fastapi.responses import JSONResponse, ORJSONResponse, UJSONResponse
from functools import lru_cache, wraps
from typing import Literal, Optional, Union
from user_agents import parse from user_agents import parse
from typing import Optional, Union, Literal from module_admin.entity.vo.log_vo import LogininforModel, OperLogModel
from module_admin.service.log_service import LoginLogService, OperationLogService
from module_admin.service.login_service import LoginService from module_admin.service.login_service import LoginService
from module_admin.service.log_service import OperationLogService, LoginLogService
from module_admin.entity.vo.log_vo import OperLogModel, LogininforModel
from config.env import AppConfig
from config.enums import BusinessType from config.enums import BusinessType
from config.env import AppConfig
def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], BusinessType], log_type: Optional[Literal['login', 'operation']] = 'operation'): def log_decorator(
title: str,
business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], BusinessType],
log_type: Optional[Literal['login', 'operation']] = 'operation',
):
""" """
日志装饰器 日志装饰器
:param log_type: 日志类型login表示登录日志为空表示为操作日志 :param log_type: 日志类型login表示登录日志为空表示为操作日志
@ -29,7 +33,11 @@ def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6,
if isinstance(business_type, BusinessType): if isinstance(business_type, BusinessType):
business_type = business_type.value business_type = business_type.value
else: else:
warnings.warn('@log_decorator的business_type参数未来版本将不再支持int类型,请使用BusinessType枚举类型', category=DeprecationWarning, stacklevel=2) warnings.warn(
'@log_decorator的business_type参数未来版本将不再支持int类型,请使用BusinessType枚举类型',
category=DeprecationWarning,
stacklevel=2,
)
def decorator(func): def decorator(func):
@wraps(func) @wraps(func)
@ -50,22 +58,24 @@ def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6,
request_method = request.method request_method = request.method
operator_type = 0 operator_type = 0
user_agent = request.headers.get('User-Agent') user_agent = request.headers.get('User-Agent')
if "Windows" in user_agent or "Macintosh" in user_agent or "Linux" in user_agent: if 'Windows' in user_agent or 'Macintosh' in user_agent or 'Linux' in user_agent:
operator_type = 1 operator_type = 1
if "Mobile" in user_agent or "Android" in user_agent or "iPhone" in user_agent: if 'Mobile' in user_agent or 'Android' in user_agent or 'iPhone' in user_agent:
operator_type = 2 operator_type = 2
# 获取请求的url # 获取请求的url
oper_url = request.url.path oper_url = request.url.path
# 获取请求的ip及ip归属区域 # 获取请求的ip及ip归属区域
oper_ip = request.headers.get("X-Forwarded-For") oper_ip = request.headers.get('X-Forwarded-For')
oper_location = '内网IP' oper_location = '内网IP'
if AppConfig.app_ip_location_query: if AppConfig.app_ip_location_query:
oper_location = get_ip_location(oper_ip) oper_location = get_ip_location(oper_ip)
# 根据不同的请求类型使用不同的方法获取请求参数 # 根据不同的请求类型使用不同的方法获取请求参数
content_type = request.headers.get("Content-Type") content_type = request.headers.get('Content-Type')
if content_type and ("multipart/form-data" in content_type or 'application/x-www-form-urlencoded' in content_type): if content_type and (
'multipart/form-data' in content_type or 'application/x-www-form-urlencoded' in content_type
):
payload = await request.form() payload = await request.form()
oper_param = "\n".join([f"{key}: {value}" for key, value in payload.items()]) oper_param = '\n'.join([f'{key}: {value}' for key, value in payload.items()])
else: else:
payload = await request.body() payload = await request.body()
# 通过 request.path_params 直接访问路径参数 # 通过 request.path_params 直接访问路径参数
@ -97,7 +107,7 @@ def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6,
loginLocation=oper_location, loginLocation=oper_location,
browser=browser, browser=browser,
os=system_os, os=system_os,
loginTime=oper_time.strftime('%Y-%m-%d %H:%M:%S') loginTime=oper_time.strftime('%Y-%m-%d %H:%M:%S'),
) )
kwargs['form_data'].login_info = login_log kwargs['form_data'].login_info = login_log
# 调用原始函数 # 调用原始函数
@ -105,10 +115,18 @@ def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6,
# 获取请求耗时 # 获取请求耗时
cost_time = float(time.time() - start_time) * 100 cost_time = float(time.time() - start_time) * 100
# 判断请求是否来自api文档 # 判断请求是否来自api文档
request_from_swagger = request.headers.get('referer').endswith('docs') if request.headers.get('referer') else False request_from_swagger = (
request_from_redoc = request.headers.get('referer').endswith('redoc') if request.headers.get('referer') else False request.headers.get('referer').endswith('docs') if request.headers.get('referer') else False
)
request_from_redoc = (
request.headers.get('referer').endswith('redoc') if request.headers.get('referer') else False
)
# 根据响应结果的类型使用不同的方法获取响应结果参数 # 根据响应结果的类型使用不同的方法获取响应结果参数
if isinstance(result, JSONResponse) or isinstance(result, ORJSONResponse) or isinstance(result, UJSONResponse): if (
isinstance(result, JSONResponse)
or isinstance(result, ORJSONResponse)
or isinstance(result, UJSONResponse)
):
result_dict = json.loads(str(result.body, 'utf-8')) result_dict = json.loads(str(result.body, 'utf-8'))
else: else:
if request_from_swagger or request_from_redoc: if request_from_swagger or request_from_redoc:
@ -160,7 +178,7 @@ def log_decorator(title: str, business_type: Union[Literal[0, 1, 2, 3, 4, 5, 6,
status=status, status=status,
errorMsg=error_msg, errorMsg=error_msg,
operTime=oper_time, operTime=oper_time,
costTime=int(cost_time) costTime=int(cost_time),
) )
await OperationLogService.add_operation_log_services(query_db, operation_log) await OperationLogService.add_operation_log_services(query_db, operation_log)

13
ruoyi-fastapi-backend/module_admin/annotation/pydantic_annotation.py

@ -1,9 +1,8 @@
import inspect import inspect
from typing import Type from fastapi import Form, Query
from fastapi import Query, Form
from pydantic import BaseModel from pydantic import BaseModel
from pydantic.fields import FieldInfo from pydantic.fields import FieldInfo
from typing import Type
def as_query(cls: Type[BaseModel]): def as_query(cls: Type[BaseModel]):
@ -21,7 +20,7 @@ def as_query(cls: Type[BaseModel]):
model_field.alias, model_field.alias,
inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_ONLY,
default=Query(model_field.default), default=Query(model_field.default),
annotation=model_field.annotation annotation=model_field.annotation,
) )
) )
else: else:
@ -30,7 +29,7 @@ def as_query(cls: Type[BaseModel]):
model_field.alias, model_field.alias,
inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_ONLY,
default=Query(...), default=Query(...),
annotation=model_field.annotation annotation=model_field.annotation,
) )
) )
@ -59,7 +58,7 @@ def as_form(cls: Type[BaseModel]):
model_field.alias, model_field.alias,
inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_ONLY,
default=Form(model_field.default), default=Form(model_field.default),
annotation=model_field.annotation annotation=model_field.annotation,
) )
) )
else: else:
@ -68,7 +67,7 @@ def as_form(cls: Type[BaseModel]):
model_field.alias, model_field.alias,
inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_ONLY,
default=Form(...), default=Form(...),
annotation=model_field.annotation annotation=model_field.annotation,
) )
) )

Loading…
Cancel
Save