新增管理员创建用户接口 POST /admin/users，创建后邮箱自动确认

app/routers/admin.py

@@ -6,6 +6,7 @@ from zoneinfo import ZoneInfo
 
 from fastapi import APIRouter, HTTPException, Query, status
 from sqlalchemy import case, func, select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload, selectinload
 
 from app.deps import AdminUser, DbSession
@@ -13,6 +14,7 @@ from app.models import Cloud, Profile, User
 from app.schemas import (
     AdminCloudStatusIn,
     AdminCloudVisibilityIn,
+    AdminCreateUserIn,
     AdminStatsOut,
     AdminUserOut,
     AdminUserUpdateIn,
@@ -22,6 +24,7 @@ from app.schemas import (
     DeleteResultOut,
     PageOut,
 )
+from app.security import hash_password, utc_now
 from app.serializers import cloud_out, profile_out, user_out
 from app.services.storage import delete_files
 
@@ -83,6 +86,32 @@ async def list_users(
     )
 
 
+@router.post("/users", response_model=AdminUserOut, status_code=status.HTTP_201_CREATED)
+async def create_user(
+    payload: AdminCreateUserIn,
+    _: AdminUser,
+    db: DbSession,
+) -> AdminUserOut:
+    email = str(payload.email).strip().lower()
+    user = User(
+        email=email,
+        password_hash=hash_password(payload.password),
+        email_verified_at=utc_now(),
+    )
+    user.profile = Profile(username=payload.username, role=payload.role)
+    db.add(user)
+    try:
+        await db.commit()
+    except IntegrityError as exc:
+        await db.rollback()
+        message = str(exc.orig).lower()
+        if "email" in message:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该邮箱已被注册") from exc
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="这个昵称已经被使用") from exc
+    await db.refresh(user, ["profile"])
+    return AdminUserOut(user=user_out(user), profile=profile_out(user.profile))
+
+
 @router.patch("/users/{user_id}", response_model=AdminUserOut)
 async def update_user(
     user_id: uuid.UUID,

app/schemas.py

@@ -242,6 +242,21 @@ class AdminUserUpdateIn(BaseModel):
         return self
 
 
+class AdminCreateUserIn(BaseModel):
+    email: EmailStr
+    password: str = Field(min_length=8, max_length=128)
+    username: str = Field(min_length=2, max_length=32)
+    role: Role = "user"
+
+    @field_validator("username")
+    @classmethod
+    def strip_username(cls, value: str) -> str:
+        value = value.strip()
+        if len(value) < 2:
+            raise ValueError("昵称至少需要 2 个字符")
+        return value
+
+
 class AdminCloudStatusIn(BatchIdsIn):
     status: CloudStatus
 

tests/test_admin.py

@@ -52,6 +52,78 @@ def test_list_users(client: TestClient) -> None:
     assert r.json()["total"] >= 1
 
 
+# ---------------------------------------------------------------------------
+# create user
+# ---------------------------------------------------------------------------
+
+
+def test_create_user(client: TestClient) -> None:
+    adm = admin_token(client)
+    r = client.post(
+        "/api/v1/admin/users",
+        headers=bearer(adm),
+        json={
+            "email": f"admin-created-{uuid.uuid4().hex[:8]}@example.com",
+            "password": "createdpass12",
+            "username": "管理员创建的用户",
+        },
+    )
+    assert r.status_code == 201
+    body = r.json()
+    assert body["user"]["email_verified"] is True
+    assert body["profile"]["role"] == "user"
+
+
+def test_create_user_with_role(client: TestClient) -> None:
+    adm = admin_token(client)
+    r = client.post(
+        "/api/v1/admin/users",
+        headers=bearer(adm),
+        json={
+            "email": f"newadmin-{uuid.uuid4().hex[:8]}@example.com",
+            "password": "newadmin12",
+            "username": "新建管理员",
+            "role": "admin",
+        },
+    )
+    assert r.status_code == 201
+    assert r.json()["profile"]["role"] == "admin"
+
+
+def test_create_user_duplicate_email(client: TestClient) -> None:
+    adm = admin_token(client)
+    email = f"dup-admin-{uuid.uuid4().hex[:8]}@example.com"
+    client.post(
+        "/api/v1/admin/users",
+        headers=bearer(adm),
+        json={"email": email, "password": "dup1pass12", "username": "第一人"},
+    )
+    r = client.post(
+        "/api/v1/admin/users",
+        headers=bearer(adm),
+        json={"email": email, "password": "dup2pass12", "username": "第二人"},
+    )
+    assert r.status_code == 409
+
+
+def test_create_user_non_admin(client: TestClient, monkeypatch) -> None:
+    info = register_and_confirm(
+        client, monkeypatch,
+        f"na-create-{uuid.uuid4().hex[:8]}@example.com",
+        "nonadmin1", "非管理员创建",
+    )
+    r = client.post(
+        "/api/v1/admin/users",
+        headers=bearer(info["token"]),
+        json={
+            "email": f"should-fail-{uuid.uuid4().hex[:8]}@example.com",
+            "password": "failpass12",
+            "username": "失败的用户",
+        },
+    )
+    assert r.status_code == 403
+
+
 # ---------------------------------------------------------------------------
 # update user
 # ---------------------------------------------------------------------------