Spaces:
Running
Running
File size: 2,425 Bytes
09801ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
Workspace Operations Module — PostgreSQL workspace membership queries.
Provides: get_workspace_members, get_user_email, is_workspace_member
"""
import uuid
from typing import Optional, List, Dict
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy import func
from database.orm import WorkspaceMember, UserProfile
async def get_workspace_members(db: AsyncSession, workspace_id: str) -> List[Dict]:
"""
Get all members of a workspace with their roles.
"""
try:
ws_id = uuid.UUID(workspace_id)
except (ValueError, AttributeError):
return []
stmt = (
select(WorkspaceMember.user_id, WorkspaceMember.role)
.where(WorkspaceMember.workspace_id == ws_id)
)
result = await db.execute(stmt)
rows = result.all()
return [{"user_id": str(row.user_id), "role": row.role} for row in rows]
async def get_user_email(db: AsyncSession, user_id: str) -> Optional[str]:
"""
Get a user's email address by their ID.
"""
try:
uid = uuid.UUID(user_id)
except (ValueError, AttributeError):
return None
stmt = select(UserProfile.email).where(UserProfile.id == uid)
result = await db.execute(stmt)
row = result.scalar_one_or_none()
return row
async def is_workspace_member(db: AsyncSession, workspace_id: str, user_id: str) -> bool:
"""
Check if a user is a member of a workspace.
"""
try:
ws_id = uuid.UUID(workspace_id)
uid = uuid.UUID(user_id)
except (ValueError, AttributeError):
return False
stmt = (
select(func.count())
.select_from(WorkspaceMember)
.where(
WorkspaceMember.workspace_id == ws_id,
WorkspaceMember.user_id == uid
)
)
result = await db.execute(stmt)
count = result.scalar()
return count > 0
async def add_workspace_member(
db: AsyncSession,
workspace_id: str,
user_id: str,
role: str = "member"
) -> Optional[WorkspaceMember]:
"""Add a user to a workspace."""
try:
member = WorkspaceMember(
workspace_id=uuid.UUID(workspace_id),
user_id=uuid.UUID(user_id),
role=role
)
db.add(member)
await db.commit()
await db.refresh(member)
return member
except Exception:
await db.rollback()
return None
|