# -*- coding: utf-8 -*-
"""Location: ./mcpgateway/services/personal_team_service.py
Copyright 2025
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti
Personal Team Service.
This module provides automatic personal team creation and management
for email-based user authentication system.
Examples:
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> isinstance(service, PersonalTeamService)
True
>>> hasattr(service, 'db')
True
"""
# Standard
from typing import Optional
# Third-Party
from sqlalchemy.orm import Session
# First-Party
from mcpgateway.db import EmailTeam, EmailTeamMember, EmailTeamMemberHistory, EmailUser, utc_now
from mcpgateway.services.logging_service import LoggingService
# Initialize logging
logging_service = LoggingService()
logger = logging_service.get_logger(__name__)
class PersonalTeamService:
"""Service for managing personal teams.
This service handles automatic creation of personal teams for users
and manages team membership for personal workspaces.
Attributes:
db (Session): SQLAlchemy database session
Examples:
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> service.__class__.__name__
'PersonalTeamService'
>>> hasattr(service, 'db')
True
"""
def __init__(self, db: Session):
"""Initialize the personal team service.
Args:
db: SQLAlchemy database session
Examples:
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> hasattr(service, 'db') and service.db is not None or service.db is None
True
"""
self.db = db
async def create_personal_team(self, user: EmailUser) -> EmailTeam:
"""Create a personal team for a user.
Args:
user: EmailUser instance for whom to create personal team
Returns:
EmailTeam: The created personal team
Raises:
ValueError: If user already has a personal team
Exception: If team creation fails
Examples:
Personal team creation is handled automatically during user registration.
The team name is derived from the user's full name or email.
After creation, a record is inserted into EmailTeamMemberHistory to track the membership event.
Note:
This method is async and cannot be directly called with 'await' in doctest. To test async methods, use an event loop in real tests.
# Example (not executable in doctest):
# import asyncio
# team = asyncio.run(service.create_personal_team(user))
"""
try:
# Check if user already has a personal team
existing_team = self.db.query(EmailTeam).filter(EmailTeam.created_by == user.email, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first()
if existing_team:
logger.warning(f"User {user.email} already has a personal team: {existing_team.id}")
raise ValueError(f"User {user.email} already has a personal team")
# Generate team name from user's display name
display_name = user.get_display_name()
team_name = f"{display_name}'s Team"
# Create team slug from email to ensure uniqueness
email_slug = user.email.replace("@", "-").replace(".", "-").lower()
team_slug = f"personal-{email_slug}"
# Create the personal team
team = EmailTeam(
name=team_name,
slug=team_slug, # Will be auto-generated by event listener if not set
description=f"Personal workspace for {user.email}",
created_by=user.email,
is_personal=True,
visibility="private",
is_active=True,
)
self.db.add(team)
self.db.flush() # Get the team ID
# Add the user as the owner of their personal team
membership = EmailTeamMember(team_id=team.id, user_email=user.email, role="owner", joined_at=utc_now(), is_active=True)
self.db.add(membership)
self.db.flush() # Get the membership ID
# Insert history record
history = EmailTeamMemberHistory(team_member_id=membership.id, team_id=team.id, user_email=user.email, role="owner", action="added", action_by=user.email, action_timestamp=utc_now())
self.db.add(history)
self.db.commit()
logger.info(f"Created personal team '{team.name}' for user {user.email}")
return team
except Exception as e:
self.db.rollback()
logger.error(f"Failed to create personal team for {user.email}: {e}")
raise
async def get_personal_team(self, user_email: str) -> Optional[EmailTeam]:
"""Get the personal team for a user.
Args:
user_email: Email address of the user
Returns:
EmailTeam: The user's personal team or None if not found
Examples:
>>> import asyncio
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> asyncio.iscoroutinefunction(service.get_personal_team)
True
"""
try:
team = self.db.query(EmailTeam).filter(EmailTeam.created_by == user_email, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first()
return team
except Exception as e:
logger.error(f"Failed to get personal team for {user_email}: {e}")
return None
async def ensure_personal_team(self, user: EmailUser) -> EmailTeam:
"""Ensure a user has a personal team, creating one if needed.
Args:
user: EmailUser instance
Returns:
EmailTeam: The user's personal team (existing or newly created)
Raises:
Exception: If team creation or retrieval fails
Examples:
>>> import asyncio
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> asyncio.iscoroutinefunction(service.ensure_personal_team)
True
"""
try:
# Try to get existing personal team
team = await self.get_personal_team(user.email)
if team is None:
# Create personal team if it doesn't exist
logger.info(f"Creating missing personal team for user {user.email}")
team = await self.create_personal_team(user)
return team
except ValueError:
# User already has a team, get it
team = await self.get_personal_team(user.email)
if team is None:
raise Exception(f"Failed to get or create personal team for {user.email}")
return team
def is_personal_team(self, team_id: str) -> bool:
"""Check if a team is a personal team.
Args:
team_id: ID of the team to check
Returns:
bool: True if the team is a personal team, False otherwise
Examples:
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> callable(service.is_personal_team)
True
"""
try:
team = self.db.query(EmailTeam).filter(EmailTeam.id == team_id, EmailTeam.is_active.is_(True)).first()
return team is not None and team.is_personal
except Exception as e:
logger.error(f"Failed to check if team {team_id} is personal: {e}")
return False
async def delete_personal_team(self, team_id: str) -> bool:
"""Delete a personal team (not allowed).
Personal teams cannot be deleted, only deactivated.
Args:
team_id: ID of the team to delete
Returns:
bool: False (personal teams cannot be deleted)
Raises:
ValueError: Always, as personal teams cannot be deleted
Examples:
>>> import asyncio
>>> from unittest.mock import Mock
>>> service = PersonalTeamService(Mock())
>>> asyncio.iscoroutinefunction(service.delete_personal_team)
True
"""
if self.is_personal_team(team_id):
raise ValueError("Personal teams cannot be deleted")
return False
async def get_personal_team_owner(self, team_id: str) -> Optional[str]:
"""Get the owner email of a personal team.
Args:
team_id: ID of the personal team
Returns:
str: Owner email address or None if not found
Examples:
Used for access control and team management operations.
"""
try:
team = self.db.query(EmailTeam).filter(EmailTeam.id == team_id, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first()
return team.created_by if team else None
except Exception as e:
logger.error(f"Failed to get personal team owner for {team_id}: {e}")
return None
