refactor(base_params): 重构分页查询参数处理逻辑

refactor(validator): 优化日期验证器并修复序列化问题

refactor(controller): 移除未使用的导入和接口

refactor(service): 优化类型提示和错误处理

fix(model): 修复JobLogModel字段定义

style(param): 移除未使用的导入和优化时间处理逻辑

fix(service): 修复服务器监控信息返回格式

fix(auth): 增强登录验证和错误处理

fix(dict): 修复字典数据状态更新逻辑

refactor(job): 优化任务服务和日志查询逻辑
This commit is contained in:
zhangtao
2025-10-01 19:29:27 +08:00
parent 2076f2e325
commit ef7143b2ae
37 changed files with 153 additions and 228 deletions
@@ -2,8 +2,8 @@
import time
from typing import Union, Dict
from fastapi import APIRouter, Depends, Request, BackgroundTasks, WebSocket
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from redis.asyncio.client import Redis
@@ -82,33 +82,3 @@ async def logout_controller(
logger.info('退出成功')
return SuccessResponse(msg='退出成功')
return ErrorResponse(msg='退出失败')
# 以下接口为预留,后期会用
@AuthRouter.post("/background_tasks", summary="模拟fastapi自带后台任务-模拟流式响应")
async def stream_response__controller(
background_tasks: BackgroundTasks,
message: str
):
def task(message):
for i in range(5):
yield f"睡眠 {message} {i}\n"
time.sleep(1)
background_tasks.add_task(task, message)
return StreamingResponse(
data=task(message),
headers={"X-Custom-Header": "Streaming-Response"},
media_type="text/plain",
)
@AuthRouter.websocket("/ws", name="websocket")
async def websocket_endpoint_controller(
websocket: WebSocket
):
# ws://127.0.0.1:8000/ws
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
@@ -64,6 +64,8 @@ class LoginService:
# 验证码校验
if settings.CAPTCHA_ENABLE and not request_from_docs:
if not login_form.captcha_key or not login_form.captcha:
raise CustomException(msg="验证码不能为空")
await CaptchaService.check_captcha_service(redis=redis, key=login_form.captcha_key, captcha=login_form.captcha)
# 用户认证
@@ -81,6 +83,10 @@ class LoginService:
# 更新最后登录时间
user = await UserCRUD(auth).update_last_login_crud(id=user.id)
if not user:
raise CustomException(msg="用户不存在")
if not login_form.login_type:
raise CustomException(msg="登录类型不能为空")
# 创建token
token = await cls.create_token_service(request=request, redis=redis, user=user, login_type=login_form.login_type)
@@ -111,7 +117,11 @@ class LoginService:
request_ip = x_forwarded_for.split(',')[0].strip()
else:
# 若没有 X-Forwarded-For 头,则使用 request.client.host
request_ip = request.client.host
if request.client:
request_ip = request.client.host
else:
request_ip = "127.0.0.1"
login_location = await IpLocalUtil.get_ip_location(request_ip)
request.scope["login_location"] = login_location
@@ -131,7 +141,7 @@ class LoginService:
login_location=login_location,
os=user_agent.os.family,
browser = user_agent.browser.family,
login_time=user.last_login.isoformat() if isinstance(user.last_login, datetime) else str(user.last_login),
login_time=user.last_login,
login_type=login_type
).model_dump_json()
@@ -162,7 +172,7 @@ class LoginService:
return JWTOutSchema(
access_token=access_token,
refresh_token=refresh_token,
expires_in=access_expires.total_seconds(),
expires_in=int(access_expires.total_seconds()),
token_type=settings.TOKEN_TYPE
)
@@ -231,7 +241,7 @@ class LoginService:
return JWTOutSchema(
access_token=access_token,
refresh_token=refresh_token_new,
expires_in=access_expires.total_seconds(),
expires_in=int(access_expires.total_seconds()),
token_type=settings.TOKEN_TYPE
)
@@ -1,15 +1,13 @@
# -*- coding: utf-8 -*-
from fastapi import APIRouter, Body, Depends, Path, Query
from fastapi import APIRouter, Body, Depends, Path
from fastapi.responses import JSONResponse
from app.common.response import SuccessResponse
from app.common.request import PaginationService
from app.core.router_class import OperationLogRoute
from app.core.dependencies import AuthPermission
from app.core.base_schema import BatchSetAvailable
from app.core.logger import logger
from app.core.base_params import PaginationQueryParam
from ..auth.schema import AuthSchema
from .param import DeptQueryParam
from .service import DeptService
@@ -27,7 +25,8 @@ async def get_dept_tree_controller(
search: DeptQueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(permissions=["system:dept:query"]))
) -> JSONResponse:
result_dict_list = await DeptService.get_dept_tree_service(search=search, auth=auth)
order_by = [{"order": "asc"}]
result_dict_list = await DeptService.get_dept_tree_service(search=search, auth=auth, order_by=order_by)
logger.info(f"查询部门树成功")
return SuccessResponse(data=result_dict_list, msg="查询部门树成功")
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -13,11 +12,10 @@ class DeptQueryParam:
self,
name: Optional[str] = Query(None, description="部门名称"),
status: Optional[bool] = Query(None, description="部门状态(True正常 False停用)"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.name = ("like", name)
@@ -27,8 +25,6 @@ class DeptQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from typing import List, Dict
from typing import List, Dict, Optional
from app.core.base_schema import BatchSetAvailable
from app.core.exceptions import CustomException
@@ -39,7 +39,7 @@ class DeptService:
return DeptOutSchema.model_validate(dept).model_dump()
@classmethod
async def get_dept_tree_service(cls, auth: AuthSchema, search: DeptQueryParam, order_by: List[Dict] = None) -> List[Dict]:
async def get_dept_tree_service(cls, auth: AuthSchema, search: Optional[DeptQueryParam]= None, order_by: Optional[List[Dict]] = None) -> List[Dict]:
"""
获取部门树形列表service
@@ -48,10 +48,6 @@ class DeptService:
:param order_by: 排序参数
:return: 部门树形列表对象
"""
if order_by:
order_by = eval(order_by)
else:
order_by = [{"order": "asc"}]
# 使用树形结构查询,预加载children关系
dept_list = await DeptCRUD(auth).get_tree_list_crud(search=search.__dict__, order_by=order_by)
# 转换为字典列表
@@ -85,8 +85,8 @@ async def delete_type_controller(
logger.info(f"删除字典类型成功: {ids}")
return SuccessResponse(msg="删除字典类型成功")
@DictRouter.patch("/type/available/setting", summary="批量修改公告状态", description="批量修改公告状态")
async def batch_set_available_obj_controller(
@DictRouter.patch("/type/available/setting", summary="批量修改字典类型状态", description="批量修改字典类型状态")
async def batch_set_available_dict_type_controller(
data: BatchSetAvailable,
auth: AuthSchema = Depends(AuthPermission(permissions=["system:dict_type:patch"]))
) -> JSONResponse:
@@ -164,7 +164,7 @@ async def delete_data_controller(
return SuccessResponse(msg="删除字典数据成功")
@DictRouter.patch("/data/available/setting", summary="批量修改字典数据状态", description="批量修改字典数据状态")
async def batch_set_available_obj_controller(
async def batch_set_available_dict_data_controller(
data: BatchSetAvailable,
auth: AuthSchema = Depends(AuthPermission(permissions=["system:dict_data:patch"]))
) -> JSONResponse:
@@ -202,4 +202,10 @@ async def get_init_dict_data_controller(
)
logger.info(f"获取初始化字典数据成功:{dict_data_query_result}")
# 确保数据是字符串类型再进行 JSON 解析
if isinstance(dict_data_query_result, bytes):
dict_data_query_result = dict_data_query_result.decode('utf-8')
elif not isinstance(dict_data_query_result, str):
dict_data_query_result = str(dict_data_query_result)
return SuccessResponse(data=json.loads(dict_data_query_result), msg="获取初始化字典数据成功")
+6 -12
View File
@@ -2,7 +2,6 @@
from typing import Optional
from fastapi import Query
from datetime import datetime
from app.core.validator import DateTimeStr
@@ -16,8 +15,8 @@ class DictTypeQueryParam:
dict_type: Optional[str] = Query(None, description="字典类型"),
status: Optional[bool] = Query(None, description="状态(1正常 0停用)"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
@@ -31,9 +30,7 @@ class DictTypeQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
class DictDataQueryParam:
@@ -45,10 +42,9 @@ class DictDataQueryParam:
dict_type: Optional[str] = Query(None, description="字典类型"),
status: Optional[bool] = Query(None, description="状态(1正常 0停用)"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.dict_label = ("like", f"%{dict_label}%") if dict_label else None
@@ -60,6 +56,4 @@ class DictDataQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
import json
from typing import Any, List, Dict
from typing import Any, List, Dict, Optional
from redis.asyncio.client import Redis
from sqlalchemy.ext.asyncio import AsyncSession
@@ -29,14 +29,8 @@ class DictTypeService:
return DictTypeOutSchema.model_validate(obj).model_dump()
@classmethod
async def get_obj_list_service(cls, auth: AuthSchema, search: DictTypeQueryParam = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
if order_by:
order_by = eval(order_by)
obj_list = None
if search:
obj_list = await DictTypeCRUD(auth).get_obj_list_crud(search=search.__dict__, order_by=order_by)
else:
obj_list = await DictTypeCRUD(auth).get_obj_list_crud()
async def get_obj_list_service(cls, auth: AuthSchema, search: Optional[DictTypeQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
obj_list = await DictTypeCRUD(auth).get_obj_list_crud(search=search.__dict__, order_by=order_by)
return [DictTypeOutSchema.model_validate(obj).model_dump() for obj in obj_list]
@classmethod
@@ -80,7 +74,6 @@ class DictTypeService:
for item in exist_obj_type_list:
item.dict_type = data.dict_type
dict_data = DictDataUpdateSchema(
id=item.id,
dict_sort=item.dict_sort,
dict_label=item.dict_label,
dict_value=item.dict_value,
@@ -179,9 +172,7 @@ class DictDataService:
return DictDataOutSchema.model_validate(obj).model_dump()
@classmethod
async def get_obj_list_service(cls, auth: AuthSchema, search: DictDataQueryParam = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
if order_by:
order_by = eval(order_by)
async def get_obj_list_service(cls, auth: AuthSchema, search: Optional[DictDataQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
obj_list = await DictDataCRUD(auth).get_obj_list_crud(search=search.__dict__, order_by=order_by)
return [DictDataOutSchema.model_validate(obj).model_dump() for obj in obj_list]
@@ -257,15 +248,17 @@ class DictDataService:
if not exist_obj:
raise CustomException(msg='更新失败,该字典数据不存在')
exist_obj = await DictDataCRUD(auth).get(dict_label=data.dict_label)
if exist_obj and exist_obj.id != id:
if not exist_obj:
raise CustomException(msg='更新失败,该字典数据不存在')
if exist_obj.id != id:
raise CustomException(msg='更新失败,数据字典数据重复')
# 如果状态变更,需要同步更新字典类型状态并刷新缓存
if exist_obj.status != data.status:
if exist_obj.status != data.status or exist_obj.dict_type != data.dict_type:
dict_type = await DictTypeCRUD(auth).get(dict_type=exist_obj.dict_type)
if dict_type:
update_data = DictTypeUpdateSchema(
id=dict_type.id,
dict_name=dict_type.dict_name,
dict_type=dict_type.dict_type,
status=data.status,
@@ -25,7 +25,10 @@ async def get_obj_list_controller(
auth: AuthSchema = Depends(AuthPermission(permissions=["system:log:query"]))
) -> JSONResponse:
""" 查询日志 """
result_dict_list = await OperationLogService.get_log_list_service(search=search, auth=auth, order_by=page.order_by)
order_by = [{"created_at": "desc"}]
if page.order_by:
order_by = page.order_by
result_dict_list = await OperationLogService.get_log_list_service(search=search, auth=auth, order_by=order_by)
result_dict = await PaginationService.paginate(data_list= result_dict_list, page_no= page.page_no, page_size = page.page_size)
logger.info(f"查询日志成功")
return SuccessResponse(data=result_dict, msg="查询日志成功")
@@ -2,7 +2,6 @@
from typing import Optional
from fastapi import Query
from datetime import datetime
from app.core.validator import DateTimeStr
@@ -17,10 +16,9 @@ class OperationLogQueryParam:
request_ip: Optional[str] = Query(None, description="请求IP"),
response_code: Optional[int] = Query(None, description="响应状态码"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.request_path = ("like", f"%{request_path}%") if request_path else None
@@ -34,6 +32,4 @@ class OperationLogQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional
from app.core.exceptions import CustomException
from app.utils.excel_util import ExcelUtil
@@ -26,12 +26,8 @@ class OperationLogService:
return log_dict
@classmethod
async def get_log_list_service(cls, auth: AuthSchema, search: OperationLogQueryParam, order_by: List[Dict] = None) -> List[Dict]:
"""获取日志列表"""
if order_by:
order_by = eval(order_by)
else:
order_by = [{"created_at": "desc"}]
async def get_log_list_service(cls, auth: AuthSchema, search: Optional[OperationLogQueryParam], order_by: Optional[List[Dict]] = None) -> List[Dict]:
"""获取日志列表"""
log_list = await OperationLogCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
log_dict_list = [OperationLogOutSchema.model_validate(log).model_dump() for log in log_list]
return log_dict_list
@@ -24,7 +24,8 @@ async def get_menu_tree_controller(
search: MenuQueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(permissions=["system:menu:query"]))
) -> JSONResponse:
result_dict_list = await MenuService.get_menu_tree_service(search=search, auth=auth)
order_by = [{"order": "asc"}]
result_dict_list = await MenuService.get_menu_tree_service(search=search, auth=auth, order_by=order_by)
logger.info(f"查询菜单树成功")
return SuccessResponse(data=result_dict_list, msg="查询菜单树成功")
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional, Literal
from fastapi import Query
@@ -17,10 +16,9 @@ class MenuQueryParam:
type: Optional[Literal['M', 'C', 'F']] = Query(None, description="菜单类型(M目录 C菜单 F按钮)"),
permission: Optional[str] = Query(None, description="权限标识"),
status: Optional[bool] = Query(None, description="菜单状态(True正常 False停用)"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.name = ("like", name)
@@ -34,7 +32,5 @@ class MenuQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from typing import List, Dict
from typing import List, Dict, Optional
from app.core.base_schema import BatchSetAvailable
from app.core.exceptions import CustomException
@@ -33,7 +33,7 @@ class MenuService:
return menu_dict
@classmethod
async def get_menu_tree_service(cls, auth: AuthSchema, search: MenuQueryParam, order_by: List[Dict] = None) -> List[Dict]:
async def get_menu_tree_service(cls, auth: AuthSchema, search: Optional[MenuQueryParam] = None, order_by: Optional[List[Dict]] = None) -> List[Dict]:
"""
获取菜单树形列表service
@@ -42,10 +42,6 @@ class MenuService:
:param order_by: 排序参数
:return: 菜单树形列表对象
"""
if order_by:
order_by = eval(order_by)
else:
order_by = [{"order": "asc"}]
# 使用树形结构查询,预加载children关系
menu_list = await MenuCRUD(auth).get_tree_list_crud(search=search.__dict__, order_by=order_by)
# 转换为字典列表
@@ -74,6 +70,8 @@ class MenuService:
if data.parent_id:
parent_menu = await MenuCRUD(auth).get_by_id_crud(id=data.parent_id)
if not parent_menu:
raise CustomException(msg='更新失败,父级菜单不存在')
data.parent_name = parent_menu.name
new_menu = await MenuCRUD(auth).update(id=id, data=data)
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -16,10 +15,9 @@ class NoticeQueryParam:
notice_type: Optional[str] = Query(None, description="公告类型"),
status: Optional[bool] = Query(None, description="是否可用"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.notice_title = ("like", notice_title)
@@ -31,8 +29,6 @@ class NoticeQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from typing import Any, List, Dict
from typing import Any, List, Dict, Optional
from app.core.base_schema import BatchSetAvailable
@@ -28,9 +28,7 @@ class NoticeService:
return [NoticeOutSchema.model_validate(notice_obj).model_dump() for notice_obj in notice_obj_list]
@classmethod
async def get_notice_list_service(cls, auth: AuthSchema, search: NoticeQueryParam = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
if order_by:
order_by = eval(order_by)
async def get_notice_list_service(cls, auth: AuthSchema, search: Optional[NoticeQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
notice_obj_list = await NoticeCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
return [NoticeOutSchema.model_validate(notice_obj).model_dump() for notice_obj in notice_obj_list]
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -14,11 +13,10 @@ class ParamsQueryParam:
config_name: Optional[str] = Query(None, description="配置名称"),
config_key: Optional[str] = Query(None, description="配置键名"),
config_type: Optional[bool] = Query(None, description="系统内置((True:是 False:否))"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.config_name = ("like", config_name)
self.config_key = ("like", config_key)
@@ -28,8 +26,6 @@ class ParamsQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -14,10 +13,9 @@ class PositionQueryParam:
name: Optional[str] = Query(None, description="岗位名称"),
status: Optional[bool] = Query(None, description="是否可用"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.name = ("like", name)
@@ -28,6 +26,4 @@ class PositionQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -14,10 +13,9 @@ class RoleQueryParam:
name: Optional[str] = Query(None, description="角色名称"),
status: Optional[bool] = Query(None, description="是否可用"),
creator: Optional[int] = Query(None, description="创建人"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
) -> None:
super().__init__()
# 模糊查询字段
self.name = ("like", name)
@@ -28,6 +26,4 @@ class RoleQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Optional
from fastapi import Query
@@ -17,8 +16,8 @@ class UserQueryParam:
email: Optional[str] = Query(None, description="邮箱", pattern=r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'),
dept_id: Optional[int] = Query(None, description="部门ID"),
status: Optional[bool] = Query(None, description="是否可用"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2025-01-01 00:00:00"),
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2025-12-31 23:59:59"),
creator: Optional[int] = Query(None, description="创建人"),
) -> None:
@@ -35,6 +34,4 @@ class UserQueryParam:
# 时间范围查询
if start_time and end_time:
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
self.created_at = ("between", (start_datetime, end_datetime))
self.created_at = ("between", (start_time, end_time))