132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import get_db
|
|
from app.deps import require_admin
|
|
from app.models import Unit, User
|
|
from app.schemas.unit import UnitCreate, UnitResponse, UnitUpdate
|
|
from app.services.audit_service import write_audit
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _serialize(unit: Unit, user_count: int) -> UnitResponse:
|
|
return UnitResponse(
|
|
id=unit.id,
|
|
name=unit.name,
|
|
description=unit.description,
|
|
parent_id=unit.parent_id,
|
|
category=unit.category or "company",
|
|
user_count=user_count,
|
|
)
|
|
|
|
|
|
def _validate_parent(
|
|
db: Session,
|
|
parent_id: int | None,
|
|
self_id: int | None,
|
|
category: str | None = None,
|
|
) -> None:
|
|
if parent_id is None:
|
|
return
|
|
if self_id is not None and parent_id == self_id:
|
|
raise HTTPException(status_code=400, detail="上级单位不能是自身")
|
|
parent = db.get(Unit, parent_id)
|
|
if not parent:
|
|
raise HTTPException(status_code=404, detail="上级单位不存在")
|
|
if category == "company" and (parent.category or "company") != "company":
|
|
raise HTTPException(status_code=400, detail="公司只能挂在公司下")
|
|
if self_id is not None:
|
|
current = parent
|
|
while current is not None:
|
|
if current.id == self_id:
|
|
raise HTTPException(status_code=400, detail="上级单位不能是自身的下级")
|
|
current = db.get(Unit, current.parent_id) if current.parent_id else None
|
|
|
|
|
|
@router.get("", response_model=list[UnitResponse])
|
|
def list_units(db: Session = Depends(get_db), _: User = Depends(require_admin)) -> list[UnitResponse]:
|
|
counts = dict(db.execute(
|
|
select(User.unit_id, func.count(User.id))
|
|
.where(User.unit_id.is_not(None))
|
|
.group_by(User.unit_id)
|
|
).all())
|
|
units = db.scalars(select(Unit).order_by(Unit.id)).all()
|
|
return [_serialize(u, counts.get(u.id, 0)) for u in units]
|
|
|
|
|
|
@router.post("", response_model=UnitResponse)
|
|
def create_unit(
|
|
payload: UnitCreate,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> UnitResponse:
|
|
if db.scalar(select(Unit).where(Unit.name == payload.name)):
|
|
raise HTTPException(status_code=409, detail="单位名称已存在")
|
|
_validate_parent(db, payload.parent_id, None, payload.category)
|
|
unit = Unit(
|
|
name=payload.name,
|
|
description=payload.description,
|
|
parent_id=payload.parent_id,
|
|
category=payload.category,
|
|
)
|
|
db.add(unit)
|
|
db.flush()
|
|
write_audit(db, "unit.create", actor_id=admin.id, target_type="unit", target_id=str(unit.id))
|
|
db.commit()
|
|
db.refresh(unit)
|
|
return _serialize(unit, 0)
|
|
|
|
|
|
@router.patch("/{unit_id}", response_model=UnitResponse)
|
|
def update_unit(
|
|
unit_id: int,
|
|
payload: UnitUpdate,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> UnitResponse:
|
|
unit = db.get(Unit, unit_id)
|
|
if not unit:
|
|
raise HTTPException(status_code=404, detail="单位不存在")
|
|
if payload.name is not None and payload.name != unit.name:
|
|
if db.scalar(select(Unit).where(Unit.name == payload.name, Unit.id != unit.id)):
|
|
raise HTTPException(status_code=409, detail="单位名称已存在")
|
|
unit.name = payload.name
|
|
if payload.description is not None:
|
|
unit.description = payload.description
|
|
fields_set = payload.model_fields_set
|
|
new_category = payload.category if payload.category is not None else (unit.category or "company")
|
|
if "parent_id" in fields_set:
|
|
_validate_parent(db, payload.parent_id, unit.id, new_category)
|
|
unit.parent_id = payload.parent_id
|
|
elif payload.category is not None and payload.category != unit.category:
|
|
# 分类变化时校验原有父级仍然合法
|
|
_validate_parent(db, unit.parent_id, unit.id, new_category)
|
|
if payload.category is not None:
|
|
unit.category = payload.category
|
|
write_audit(db, "unit.update", actor_id=admin.id, target_type="unit", target_id=str(unit.id))
|
|
db.commit()
|
|
db.refresh(unit)
|
|
user_count = db.scalar(select(func.count(User.id)).where(User.unit_id == unit.id)) or 0
|
|
return _serialize(unit, user_count)
|
|
|
|
|
|
@router.delete("/{unit_id}")
|
|
def delete_unit(
|
|
unit_id: int,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> dict:
|
|
unit = db.get(Unit, unit_id)
|
|
if not unit:
|
|
raise HTTPException(status_code=404, detail="单位不存在")
|
|
if db.scalar(select(func.count(Unit.id)).where(Unit.parent_id == unit_id)):
|
|
raise HTTPException(status_code=400, detail="请先删除或转移下级单位")
|
|
db.execute(update(User).where(User.unit_id == unit_id).values(unit_id=None))
|
|
write_audit(db, "unit.delete", actor_id=admin.id, target_type="unit", target_id=str(unit_id),
|
|
detail={"name": unit.name})
|
|
db.delete(unit)
|
|
db.commit()
|
|
return {"deleted": True, "id": unit_id}
|