Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""用户管理 API"""
|
|
import hashlib
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_role, require_auth
|
|
from app.models import User
|
|
|
|
router = APIRouter(prefix="/api/cma/users", tags=["用户管理"],
|
|
dependencies=[Depends(require_role("ceo", "it"))],
|
|
)
|
|
|
|
def user_to_dict(u):
|
|
return {
|
|
"id": u.id,
|
|
"username": u.username,
|
|
"name": u.name,
|
|
"role": u.role,
|
|
"phone": u.phone,
|
|
"created_at": u.created_at.isoformat() if u.created_at else None,
|
|
}
|
|
|
|
@router.get("")
|
|
def list_users(db: Session = Depends(get_db)):
|
|
users = db.query(User).order_by(User.id).all()
|
|
return {"data": [user_to_dict(u) for u in users]}
|
|
|
|
@router.post("")
|
|
def create_user(data: dict, db: Session = Depends(get_db)):
|
|
exist = db.query(User).filter(User.username == data.get("username")).first()
|
|
if exist:
|
|
raise HTTPException(400, "用户名已存在")
|
|
user = User(
|
|
username=data["username"],
|
|
password_hash=hashlib.sha256(data["password"].encode()).hexdigest(),
|
|
name=data.get("name", data["username"]),
|
|
role=data.get("role", "business"),
|
|
phone=data.get("phone", ""),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user_to_dict(user)
|
|
|
|
@router.put("/{user_id}")
|
|
def update_user(user_id: int, data: dict, db: Session = Depends(get_db)):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(404, "用户不存在")
|
|
for k, v in data.items():
|
|
if k == "password" and v:
|
|
setattr(user, "password_hash", hashlib.sha256(v.encode()).hexdigest())
|
|
elif hasattr(user, k) and v is not None and k not in ("id", "username", "created_at"):
|
|
setattr(user, k, v)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user_to_dict(user)
|
|
|
|
@router.delete("/{user_id}")
|
|
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(404, "用户不存在")
|
|
db.delete(user)
|
|
db.commit()
|
|
return {"message": "已删除"}
|