import json from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select, delete as sa_delete, func as sa_func from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.security import get_current_user, require_role from app.models.user import User from app.models.instance import Instance from app.models.instance_group import InstanceGroup, user_groups, instance_group_members from app.schemas.instance_group import ( InstanceGroupCreate, InstanceGroupUpdate, InstanceGroupResponse, InstanceGroupDetailResponse, InstanceAssignRequest, InstanceRemoveRequest, UserAssignRequest, UserGroupUpdateRequest, ) router = APIRouter(prefix="/instance-groups", tags=["权限组管理"]) INSTANCE_FIELDS = ["id", "name", "db_type", "host", "port", "env", "status"] def _serialize_instance(inst: Instance) -> dict: return {k: getattr(inst, k, None) for k in INSTANCE_FIELDS} async def _count_members(db: AsyncSession, group_id: int, access_type: str) -> int: result = await db.execute( select(sa_func.count()).select_from(instance_group_members).where( instance_group_members.c.group_id == group_id, instance_group_members.c.access_type == access_type, ) ) return result.scalar() or 0 async def _get_member_instances(db: AsyncSession, group_id: int, access_type: str) -> List[dict]: """查询权限组中指定 access_type 的实例列表。""" # Join instance_group_members with instances result = await db.execute( select(Instance, instance_group_members.c.access_type) .join(instance_group_members, Instance.id == instance_group_members.c.instance_id) .where( instance_group_members.c.group_id == group_id, instance_group_members.c.access_type == access_type, ) ) rows = result.all() return [_serialize_instance(row[0]) for row in rows] async def _serialize_group(group: InstanceGroup, db: AsyncSession) -> dict: ro_count = await _count_members(db, group.id, "readonly") wr_count = await _count_members(db, group.id, "write") user_count = len(group.users) if group.users else 0 return { "id": group.id, "name": group.name, "description": group.description, "db_type": group.db_type, "permissions": group.get_permissions(), "allowed_databases": group.get_allowed_databases(), "is_active": group.is_active, "created_at": group.created_at, "updated_at": group.updated_at, "readonly_instance_count": ro_count, "write_instance_count": wr_count, "user_count": user_count, } async def _serialize_group_detail(group: InstanceGroup, db: AsyncSession) -> dict: d = await _serialize_group(group, db) d["readonly_instances"] = await _get_member_instances(db, group.id, "readonly") d["write_instances"] = await _get_member_instances(db, group.id, "write") d["users"] = [ { "id": u.id, "username": u.username, "display_name": u.display_name, "role": u.role, } for u in (group.users or []) ] return d async def _sync_members( db: AsyncSession, group_id: int, readonly_instance_ids: List[int], write_instance_ids: List[int], ): """Clear and rewrite all instance_group_members for a group.""" await db.execute( sa_delete(instance_group_members).where( instance_group_members.c.group_id == group_id ) ) for inst_id in readonly_instance_ids: await db.execute( instance_group_members.insert().values( group_id=group_id, instance_id=inst_id, access_type="readonly", ) ) for inst_id in write_instance_ids: await db.execute( instance_group_members.insert().values( group_id=group_id, instance_id=inst_id, access_type="write", ) ) @router.get("", response_model=List[InstanceGroupDetailResponse]) async def list_instance_groups( db_type: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): stmt = select(InstanceGroup).order_by(InstanceGroup.id) if db_type: stmt = stmt.where(InstanceGroup.db_type == db_type) result = await db.execute(stmt) groups = result.scalars().all() return [await _serialize_group_detail(g, db) for g in groups] @router.post("", response_model=InstanceGroupDetailResponse, status_code=201) async def create_instance_group( data: InstanceGroupCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): # Check name uniqueness existing = await db.execute( select(InstanceGroup).where(InstanceGroup.name == data.name) ) if existing.scalar_one_or_none(): raise HTTPException(status_code=400, detail="权限组名称已存在") group = InstanceGroup( name=data.name, description=data.description, db_type=data.db_type, allowed_databases=( json.dumps(data.allowed_databases, ensure_ascii=False) if data.allowed_databases else None ), permissions=json.dumps(data.permissions, ensure_ascii=False) if data.permissions else None, is_active=data.is_active, ) db.add(group) await db.commit() await db.refresh(group) # Write members await _sync_members(db, group.id, data.readonly_instance_ids, data.write_instance_ids) await db.commit() return await _serialize_group_detail(group, db) @router.get("/{group_id}", response_model=InstanceGroupDetailResponse) async def get_instance_group( group_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") return await _serialize_group_detail(group, db) @router.put("/{group_id}", response_model=InstanceGroupDetailResponse) async def update_instance_group( group_id: int, data: InstanceGroupUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") update_data = data.model_dump(exclude_unset=True) if "name" in update_data: # Check uniqueness existing = await db.execute( select(InstanceGroup).where( InstanceGroup.name == update_data["name"], InstanceGroup.id != group_id, ) ) if existing.scalar_one_or_none(): raise HTTPException(status_code=400, detail="权限组名称已存在") # Handle legacy permissions field if "permissions" in update_data: update_data["permissions"] = ( json.dumps(update_data["permissions"], ensure_ascii=False) if update_data["permissions"] else None ) # Handle new allowed_databases field if "allowed_databases" in update_data: update_data["allowed_databases"] = ( json.dumps(update_data["allowed_databases"], ensure_ascii=False) if update_data["allowed_databases"] else None ) # Extract member lists before setting attributes ro_ids = update_data.pop("readonly_instance_ids", None) wr_ids = update_data.pop("write_instance_ids", None) for key, value in update_data.items(): setattr(group, key, value) await db.commit() await db.refresh(group) # Update members if provided if ro_ids is not None or wr_ids is not None: await _sync_members( db, group_id, ro_ids if ro_ids is not None else [], wr_ids if wr_ids is not None else [], ) await db.commit() return await _serialize_group_detail(group, db) @router.delete("/{group_id}") async def delete_instance_group( group_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") # Delete members first await db.execute( sa_delete(instance_group_members).where( instance_group_members.c.group_id == group_id ) ) await db.delete(group) await db.commit() return {"message": "权限组已删除"} @router.post("/{group_id}/instances") async def assign_instances( group_id: int, data: InstanceAssignRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") for inst_id in data.instance_ids: inst_result = await db.execute( select(Instance).where(Instance.id == inst_id) ) inst = inst_result.scalar_one_or_none() if inst: # Check for duplicate existing = await db.execute( select(instance_group_members).where( instance_group_members.c.group_id == group_id, instance_group_members.c.instance_id == inst_id, instance_group_members.c.access_type == data.access_type, ) ) if not existing.first(): await db.execute( instance_group_members.insert().values( group_id=group_id, instance_id=inst_id, access_type=data.access_type, ) ) await db.commit() return {"message": "实例已关联"} @router.delete("/{group_id}/instances/{instance_id}") async def remove_instance( group_id: int, instance_id: int, access_type: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): conditions = [ instance_group_members.c.group_id == group_id, instance_group_members.c.instance_id == instance_id, ] if access_type: conditions.append(instance_group_members.c.access_type == access_type) result = await db.execute( sa_delete(instance_group_members).where(*conditions) ) if result.rowcount == 0: raise HTTPException(status_code=404, detail="实例未关联该权限组") await db.commit() return {"message": "实例已取消关联"} @router.get("/{group_id}/users") async def list_group_users( group_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") # Query user_groups for permissions override ug_result = await db.execute( select(user_groups).where(user_groups.c.group_id == group_id) ) ug_rows = ug_result.all() perm_map = {} for row in ug_rows: perms = json.loads(row.permissions) if row.permissions else None perm_map[row.user_id] = perms users_data = [] for u in (group.users or []): users_data.append({ "id": u.id, "username": u.username, "display_name": u.display_name, "role": u.role, "permissions": perm_map.get(u.id), }) return users_data @router.post("/{group_id}/users") async def assign_users( group_id: int, data: UserAssignRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == group_id) ) group = result.scalar_one_or_none() if not group: raise HTTPException(status_code=404, detail="权限组不存在") perm_json = json.dumps(data.permissions, ensure_ascii=False) if data.permissions else None for user_id in data.user_ids: # Check if already assigned check = await db.execute( select(user_groups).where( user_groups.c.user_id == user_id, user_groups.c.group_id == group_id, ) ) if not check.first(): await db.execute( user_groups.insert().values( user_id=user_id, group_id=group_id, permissions=perm_json, ) ) await db.commit() return {"message": "用户已分配"} @router.put("/users/{user_id}/groups") async def update_user_groups( user_id: int, data: UserGroupUpdateRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin")), ): user_result = await db.execute( select(User).where(User.id == user_id) ) user = user_result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="用户不存在") # Remove existing group associations await db.execute( user_groups.delete().where(user_groups.c.user_id == user_id) ) # Add new ones for gid in data.group_ids: group_result = await db.execute( select(InstanceGroup).where(InstanceGroup.id == gid) ) if group_result.scalar_one_or_none(): await db.execute( user_groups.insert().values(user_id=user_id, group_id=gid) ) await db.commit() return {"message": "用户权限组已更新"}