Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import secrets | |
| import hashlib | |
| from typing import Any | |
| from app.projects.repositories.collaboration_repository import CollaborationRepository | |
| from app.projects.schemas.collaboration import TeamResponse, InvitationResponse, MemberResponse | |
| from app.security.models import WorkspaceMembership | |
| from app.projects.errors import ( | |
| CollaborationUnauthorizedError, | |
| TeamNotFoundError, | |
| TeamMemberNotFoundError, | |
| ) | |
| class CollaborationService: | |
| def __init__(self, repository: CollaborationRepository) -> None: | |
| self.repository = repository | |
| async def list_teams(self, workspace_id: str) -> list[TeamResponse]: | |
| teams = await self.repository.list_teams(workspace_id) | |
| return [TeamResponse(id=t.id, workspace_id=t.workspace_id, name=t.name, created_at=t.created_at.isoformat()) for t in teams] | |
| async def create_team(self, workspace_id: str, name: str) -> TeamResponse: | |
| team = await self.repository.create_team(workspace_id, name) | |
| return TeamResponse(id=team.id, workspace_id=team.workspace_id, name=team.name, created_at=team.created_at.isoformat()) | |
| async def update_team(self, workspace_id: str, team_id: str, name: str) -> TeamResponse: | |
| try: | |
| team = await self.repository.update_team(workspace_id, team_id, name) | |
| except Exception: | |
| raise TeamNotFoundError(f"Team {team_id} not found.") | |
| return TeamResponse(id=team.id, workspace_id=team.workspace_id, name=team.name, created_at=team.created_at.isoformat()) | |
| async def archive_team(self, workspace_id: str, team_id: str) -> None: | |
| await self.repository.archive_team(workspace_id, team_id) | |
| async def invite_member(self, workspace_id: str, email: str, role: str) -> InvitationResponse: | |
| if role not in {"admin", "member"}: | |
| raise ValueError("Invitation role must be admin or member.") | |
| token = secrets.token_urlsafe(32) | |
| token_hash = hashlib.sha256(token.encode()).hexdigest() | |
| invitation = await self.repository.invite_member_with_hash(workspace_id, email, role, token_hash) | |
| response = InvitationResponse( | |
| id=invitation.id, | |
| workspace_id=invitation.workspace_id, | |
| email=invitation.email, | |
| role=invitation.role, | |
| status=invitation.status, | |
| expires_at=invitation.expires_at.isoformat(), | |
| created_at=invitation.created_at.isoformat() | |
| ) | |
| setattr(response, "token", token) | |
| return response | |
| async def list_members(self, workspace_id: str) -> list[MemberResponse]: | |
| members = await self.repository.list_members(workspace_id) | |
| return [MemberResponse(id=m.id, workspace_id=m.workspace_id, user_id=m.user_id, role=m.role, created_at=m.created_at.isoformat()) for m in members] | |
| async def list_invitations(self, workspace_id: str) -> list[InvitationResponse]: | |
| invitations = await self.repository.list_invitations(workspace_id) | |
| return [InvitationResponse( | |
| id=i.id, | |
| workspace_id=i.workspace_id, | |
| email=i.email, | |
| role=i.role, | |
| status=i.status, | |
| expires_at=i.expires_at.isoformat(), | |
| created_at=i.created_at.isoformat() | |
| ) for i in invitations] | |
| async def remove_member(self, workspace_id: str, actor_user_id: str, target_user_id: str) -> None: | |
| # Check permissions: actor must be admin | |
| actor_membership = await self.repository.get_membership(workspace_id, actor_user_id) | |
| if not actor_membership or actor_membership.role != 'admin': | |
| raise CollaborationUnauthorizedError("Unauthorized: Only admins can remove members.") | |
| # Prevent removing last admin | |
| if actor_membership.user_id == target_user_id: | |
| raise CollaborationUnauthorizedError("Cannot remove yourself.") | |
| target_membership = await self.repository.get_membership(workspace_id, target_user_id) | |
| if not target_membership: | |
| raise TeamMemberNotFoundError("Member not found.") | |
| if target_membership.role == 'admin': | |
| admin_count = await self.repository.count_workspace_admins(workspace_id) | |
| if admin_count <= 1: | |
| raise CollaborationUnauthorizedError("Cannot remove the last administrator.") | |
| # Perform removal | |
| await self.repository.remove_member(workspace_id, target_user_id) | |
| async def update_member_role(self, workspace_id: str, actor_user_id: str, target_user_id: str, new_role: str) -> None: | |
| # Check permissions: actor must be admin | |
| actor_membership = await self.repository.get_membership(workspace_id, actor_user_id) | |
| if not actor_membership or actor_membership.role != 'admin': | |
| raise CollaborationUnauthorizedError("Unauthorized: Only admins can update member roles.") | |
| # Prevent self-elevation | |
| if actor_user_id == target_user_id and new_role == 'admin': | |
| raise CollaborationUnauthorizedError("Cannot elevate your own privileges.") | |
| # Perform update | |
| await self.repository.update_member_role(workspace_id, target_user_id, new_role) | |
| async def get_membership(self, workspace_id: str, user_id: str) -> WorkspaceMembership | None: | |
| return await self.repository.get_membership(workspace_id, user_id) | |
| async def list_project_collaborators(self, project_id: str) -> list[MemberResponse]: | |
| collaborators = await self.repository.list_project_collaborators(project_id) | |
| return [MemberResponse(id=c.id, workspace_id=c.workspace_id, user_id=c.user_id, role=c.role, created_at=c.created_at.isoformat()) for c in collaborators] | |
| async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> MemberResponse: | |
| collaborator = await self.repository.add_project_collaborator(workspace_id, project_id, user_id, role) | |
| return MemberResponse(id=collaborator.id, workspace_id=collaborator.workspace_id, user_id=collaborator.user_id, role=collaborator.role, created_at=collaborator.created_at.isoformat()) | |
| async def record_activity(self, workspace_id: str, user_id: str, action: str, entity_id: str, entity_type: str, metadata: dict[str, Any]) -> None: | |
| await self.repository.record_activity(workspace_id, user_id, action, entity_id, entity_type, metadata) | |
| async def list_activity(self, workspace_id: str) -> list[dict[str, Any]]: | |
| activities = await self.repository.list_activity(workspace_id) | |
| return [ | |
| { | |
| "id": a.id, | |
| "user_id": a.user_id, | |
| "action": a.action, | |
| "entity_id": a.entity_id, | |
| "entity_type": a.entity_type, | |
| "metadata": a.metadata, | |
| "created_at": a.created_at.isoformat(), | |
| } | |
| for a in activities | |
| ] | |