65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
|
"""用户列表 — 对接 MOM sys_user"""
|
||
|
|
from fastapi import APIRouter, Query, HTTPException, status
|
||
|
|
from pydantic import BaseModel
|
||
|
|
from app.core.mom_database import MomSessionLocal
|
||
|
|
from sqlalchemy import text
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/users", tags=["用户"])
|
||
|
|
|
||
|
|
|
||
|
|
class UserOption(BaseModel):
|
||
|
|
id: str
|
||
|
|
username: str
|
||
|
|
full_name: str
|
||
|
|
|
||
|
|
model_config = {"from_attributes": True}
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/", response_model=list[UserOption])
|
||
|
|
def list_users(
|
||
|
|
keyword: str = Query("", description="按用户名/姓名模糊搜索"),
|
||
|
|
limit: int = Query(100, ge=1, le=500),
|
||
|
|
):
|
||
|
|
"""获取 MOM 系统用户列表，供前端选人使用"""
|
||
|
|
db = MomSessionLocal()
|
||
|
|
try:
|
||
|
|
if keyword.strip():
|
||
|
|
sql = text(
|
||
|
|
"""
|
||
|
|
SELECT id, username,
|
||
|
|
SPLIT_PART(username, '/', 1) AS full_name
|
||
|
|
FROM sys_user
|
||
|
|
WHERE username ILIKE :kw
|
||
|
|
ORDER BY username
|
||
|
|
LIMIT :lim
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
rows = db.execute(sql, {"kw": f"%{keyword.strip()}%", "lim": limit}).fetchall()
|
||
|
|
else:
|
||
|
|
sql = text(
|
||
|
|
"""
|
||
|
|
SELECT id, username,
|
||
|
|
SPLIT_PART(username, '/', 1) AS full_name
|
||
|
|
FROM sys_user
|
||
|
|
ORDER BY username
|
||
|
|
LIMIT :lim
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
rows = db.execute(sql, {"lim": limit}).fetchall()
|
||
|
|
|
||
|
|
return [
|
||
|
|
UserOption(
|
||
|
|
id=str(row.id),
|
||
|
|
username=row.username.split("/")[-1] if "/" in row.username else row.username,
|
||
|
|
full_name=row.full_name,
|
||
|
|
)
|
||
|
|
for row in rows
|
||
|
|
]
|
||
|
|
except Exception as e:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
|
|
detail=f"MOM 用户查询失败: {str(e)}",
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
db.close()
|