mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 13:05:18 +00:00
zhangtao
ef7143b2ae
refactor(validator): 优化日期验证器并修复序列化问题 refactor(controller): 移除未使用的导入和接口 refactor(service): 优化类型提示和错误处理 fix(model): 修复JobLogModel字段定义 style(param): 移除未使用的导入和优化时间处理逻辑 fix(service): 修复服务器监控信息返回格式 fix(auth): 增强登录验证和错误处理 fix(dict): 修复字典数据状态更新逻辑 refactor(job): 优化任务服务和日志查询逻辑
85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import time
|
|
from typing import Union, Dict
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from redis.asyncio.client import Redis
|
|
|
|
from app.common.response import ErrorResponse, SuccessResponse
|
|
from app.core.router_class import OperationLogRoute
|
|
from app.core.security import CustomOAuth2PasswordRequestForm
|
|
from app.core.logger import logger
|
|
from app.config.setting import settings
|
|
from app.core.dependencies import (
|
|
db_getter,
|
|
get_current_user,
|
|
redis_getter
|
|
)
|
|
from .service import (
|
|
LoginService,
|
|
CaptchaService
|
|
)
|
|
from .schema import (
|
|
CaptchaOutSchema,
|
|
JWTOutSchema,
|
|
RefreshTokenPayloadSchema,
|
|
LogoutPayloadSchema
|
|
)
|
|
|
|
|
|
AuthRouter = APIRouter(route_class=OperationLogRoute, prefix="/auth", tags=["认证授权"])
|
|
|
|
|
|
@AuthRouter.post("/login", summary="登录", description="登录", response_model=JWTOutSchema)
|
|
async def login_for_access_token_controller(
|
|
request: Request,
|
|
redis: Redis = Depends(redis_getter),
|
|
login_form: CustomOAuth2PasswordRequestForm = Depends(),
|
|
db: AsyncSession = Depends(db_getter),
|
|
) -> Union[JSONResponse, Dict]:
|
|
login_token = await LoginService.authenticate_user_service(request=request, redis=redis, login_form=login_form, db=db)
|
|
|
|
logger.info(f"用户{login_form.username}登录成功")
|
|
|
|
# 如果是文档请求，则不记录日志:http://localhost:8000/api/v1/docs
|
|
if settings.DOCS_URL in request.headers.get("referer", ""):
|
|
return login_token.model_dump()
|
|
return SuccessResponse(data=login_token.model_dump(), msg="登录成功")
|
|
|
|
|
|
@AuthRouter.post("/token/refresh", summary="刷新token", description="刷新token", response_model=JWTOutSchema, dependencies=[Depends(get_current_user)])
|
|
async def get_new_token_controller(
|
|
request: Request,
|
|
payload: RefreshTokenPayloadSchema,
|
|
db: AsyncSession = Depends(db_getter),
|
|
redis: Redis = Depends(redis_getter)
|
|
) -> JSONResponse:
|
|
# 解析当前的访问Token以获取用户名
|
|
new_token = await LoginService.refresh_token_service(db=db, request=request, redis=redis, refresh_token=payload)
|
|
token_dict = new_token.model_dump()
|
|
logger.info(f"刷新token成功: {token_dict}")
|
|
return SuccessResponse(data=token_dict, msg="刷新成功")
|
|
|
|
|
|
@AuthRouter.get("/captcha/get", summary="获取验证码", description="获取登录验证码", response_model=CaptchaOutSchema)
|
|
async def get_captcha_for_login_controller(
|
|
redis: Redis = Depends(redis_getter)
|
|
) -> JSONResponse:
|
|
# 获取验证码
|
|
captcha = await CaptchaService.get_captcha_service(redis=redis)
|
|
logger.info(f"获取验证码成功")
|
|
return SuccessResponse(data=captcha, msg="获取验证码成功")
|
|
|
|
|
|
@AuthRouter.post('/logout', summary="退出登录", description="退出登录", dependencies=[Depends(get_current_user)])
|
|
async def logout_controller(
|
|
payload: LogoutPayloadSchema,
|
|
redis: Redis = Depends(redis_getter)
|
|
) -> JSONResponse:
|
|
if await LoginService.logout_service(redis=redis, token=payload):
|
|
logger.info('退出成功')
|
|
return SuccessResponse(msg='退出成功')
|
|
return ErrorResponse(msg='退出失败')
|