73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
"""Password, development, and DingTalk authentication endpoints."""
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from app.api.dependencies import DatabaseSession, get_current_user
|
|
from app.core.config import get_settings
|
|
from app.core.exceptions import AppException
|
|
from app.core.security import create_access_token, hash_password
|
|
from app.models.entities import OperationAudit, UserAccount
|
|
from app.repositories.users import UserRepository
|
|
from app.schemas.auth import DingTalkConfigData, DingTalkLoginRequest, LoginRequest, TokenData, UserData
|
|
from app.schemas.common import ApiResponse
|
|
from app.services.auth import AuthService
|
|
from app.services.dingtalk_auth import DingTalkAuthService
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/login", response_model=ApiResponse[TokenData])
|
|
async def login(request: LoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]:
|
|
return ApiResponse(data=await AuthService(session).login(request.username, request.password))
|
|
|
|
|
|
@router.get("/me", response_model=ApiResponse[UserData])
|
|
async def me(
|
|
current_user: Annotated[UserAccount, Depends(get_current_user)],
|
|
session: DatabaseSession,
|
|
) -> ApiResponse[UserData]:
|
|
return ApiResponse(data=await AuthService(session).me(current_user))
|
|
|
|
|
|
@router.get("/dingtalk/config", response_model=ApiResponse[DingTalkConfigData])
|
|
async def dingtalk_config() -> ApiResponse[DingTalkConfigData]:
|
|
return ApiResponse(data=DingTalkAuthService.config())
|
|
|
|
|
|
@router.post("/dingtalk/login", response_model=ApiResponse[TokenData])
|
|
async def dingtalk_login(request: DingTalkLoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]:
|
|
return ApiResponse(data=await DingTalkAuthService(session).login(request.auth_code))
|
|
|
|
|
|
@router.post("/dev-login", response_model=ApiResponse[TokenData])
|
|
async def dev_login(request: LoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]:
|
|
"""Development-only login helper used by local frontend verification."""
|
|
|
|
settings = get_settings()
|
|
if settings.environment != "development":
|
|
raise AppException("开发登录入口未启用", code=404, status_code=404)
|
|
user = await UserRepository(session).get_by_dingtalk_id(f"local-{request.username}")
|
|
if user is None:
|
|
user = UserAccount(
|
|
emp_no=f"dev_{request.username}"[:20],
|
|
name="开发测试用户",
|
|
role="admin",
|
|
status="启用",
|
|
dingtalk_id=f"local-{request.username}",
|
|
password_hash=hash_password(request.password),
|
|
)
|
|
session.add(user)
|
|
await session.flush()
|
|
user.last_login_at = datetime.now(UTC)
|
|
session.add(OperationAudit(actor_user_id=user.id, actor_emp_no=user.emp_no, action="auth.dev_login"))
|
|
return ApiResponse(
|
|
data=TokenData(
|
|
access_token=create_access_token(user.id, role=user.role),
|
|
expires_in=settings.jwt_expire_minutes * 60,
|
|
user=await AuthService(session).me(user),
|
|
)
|
|
)
|