115 lines
3.0 KiB
Python
115 lines
3.0 KiB
Python
|
"""
|
|||
|
|
工作区 (Workspace) 服务层 — 权限检查、成员管理
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
from typing import List, Optional
|
|||
|
|
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
from app.models.user import User
|
|||
|
|
from app.models.workspace import Workspace, WorkspaceMembership
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def check_workspace_access(
|
|||
|
|
db: Session,
|
|||
|
|
user: User,
|
|||
|
|
workspace_id: str,
|
|||
|
|
required_role: str = "member",
|
|||
|
|
) -> bool:
|
|||
|
|
"""检查用户是否有工作区访问权限。
|
|||
|
|
|
|||
|
|
- 平台管理员（user.role == "admin"）总是有权限
|
|||
|
|
- required_role="member": 任意成员即可
|
|||
|
|
- required_role="admin": 必须是工作区管理员
|
|||
|
|
"""
|
|||
|
|
if user.role == "admin":
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
membership = (
|
|||
|
|
db.query(WorkspaceMembership)
|
|||
|
|
.filter(
|
|||
|
|
WorkspaceMembership.workspace_id == workspace_id,
|
|||
|
|
WorkspaceMembership.user_id == user.id,
|
|||
|
|
)
|
|||
|
|
.first()
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if not membership:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
if required_role == "member":
|
|||
|
|
return True
|
|||
|
|
elif required_role == "admin":
|
|||
|
|
return membership.role == "admin"
|
|||
|
|
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_user_workspace_ids(db: Session, user: User) -> List[str]:
|
|||
|
|
"""获取用户所属的所有工作区 ID 列表。平台管理员可看到所有活跃工作区。"""
|
|||
|
|
if user.role == "admin":
|
|||
|
|
workspaces = (
|
|||
|
|
db.query(Workspace.id)
|
|||
|
|
.filter(Workspace.status == "active")
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
return [w.id for w in workspaces]
|
|||
|
|
|
|||
|
|
memberships = (
|
|||
|
|
db.query(WorkspaceMembership.workspace_id)
|
|||
|
|
.filter(WorkspaceMembership.user_id == user.id)
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
return [m.workspace_id for m in memberships]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_user_workspaces(db: Session, user: User) -> List[dict]:
|
|||
|
|
"""获取用户的工作区列表，带角色信息。"""
|
|||
|
|
if user.role == "admin":
|
|||
|
|
workspaces = (
|
|||
|
|
db.query(Workspace)
|
|||
|
|
.filter(Workspace.status == "active")
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
return [
|
|||
|
|
{
|
|||
|
|
"id": w.id,
|
|||
|
|
"name": w.name,
|
|||
|
|
"description": w.description,
|
|||
|
|
"is_default": bool(w.is_default),
|
|||
|
|
"owner_id": w.owner_id,
|
|||
|
|
"role": "admin",
|
|||
|
|
"status": w.status,
|
|||
|
|
"created_at": w.created_at.isoformat() if w.created_at else None,
|
|||
|
|
}
|
|||
|
|
for w in workspaces
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
memberships = (
|
|||
|
|
db.query(WorkspaceMembership)
|
|||
|
|
.filter(WorkspaceMembership.user_id == user.id)
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
result = []
|
|||
|
|
for m in memberships:
|
|||
|
|
w = m.workspace
|
|||
|
|
if w.status != "active":
|
|||
|
|
continue
|
|||
|
|
result.append({
|
|||
|
|
"id": w.id,
|
|||
|
|
"name": w.name,
|
|||
|
|
"description": w.description,
|
|||
|
|
"is_default": bool(w.is_default),
|
|||
|
|
"owner_id": w.owner_id,
|
|||
|
|
"role": m.role,
|
|||
|
|
"status": w.status,
|
|||
|
|
"created_at": w.created_at.isoformat() if w.created_at else None,
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
return result
|