User Invitations

Core Components

# Invitation model definition
class UserInvitation:
    id: str
    email: str
    invited_by: str
    invitation_type: str
    role: Optional[str]
    team_id: Optional[str]
    project_id: Optional[str]
    organization_id: Optional[str]
    token: str
    status: str
    created_at: datetime
    expires_at: datetime
    accepted_at: Optional[datetime]
    message: Optional[str]
    custom_data: Optional[Dict]
class InvitationType:
    USER_REGISTRATION = "user_registration"
    TEAM_INVITATION = "team_invitation"
    PROJECT_INVITATION = "project_invitation"
    ORGANIZATION_INVITATION = "organization_invitation"
class InvitationStatus:
    PENDING = "pending"
    ACCEPTED = "accepted"
    DECLINED = "declined"
    EXPIRED = "expired"
    CANCELLED = "cancelled"

1Create Invitation

// Create user invitation
const invitation = await InvitationService.create({
  email: 'user@example.com',
  type: 'team_invitation',
  role: 'member',
  teamId: 'team-123',
  invitedBy: 'admin-user',
  message: 'Join our analytics team!',
  customData: {
    department: 'analytics',
    location: 'New York'
  }
});
console.log('Invitation created:', invitation.id);

2Send Invitation

// Send invitation email
await EmailService.sendInvitation({
  to: invitation.email,
  invitationId: invitation.id,
  token: invitation.token,
  template: 'team_invitation',
  customData: invitation.customData
});

3Track Invitation Status

// Track invitation status
const invitationStatus = await InvitationService.getStatus(invitation.id);
switch (invitationStatus.status) {
  case 'pending':
    console.log('Invitation is pending');
    break;
  case 'accepted':
    console.log('Invitation was accepted');
    break;
  case 'declined':
    console.log('Invitation was declined');
    break;
  case 'expired':
    console.log('Invitation has expired');
    break;
}

Team Invitation Flow

# Team invitation implementation
class TeamInvitationService:
    def __init__(self, team_id):
        self.team_id = team_id
    
    async def create_invitation(self, email, role='member', **kwargs):
        """Create team invitation"""
        invitation = UserInvitation(
            email=email,
            invitation_type='team_invitation',
            role=role,
            team_id=self.team_id,
            **kwargs
        )
        
        # Generate secure token
        invitation.token = self.generate_token()
        
        # Set expiration (7 days default)
        invitation.expires_at = datetime.utcnow() + timedelta(days=7)
        
        # Save invitation
        await self.save_invitation(invitation)
        
        # Send invitation email
        await self.send_team_invitation_email(invitation)
        
        return invitation
    
    async def accept_invitation(self, token, user_id):
        """Accept team invitation"""
        invitation = await self.get_invitation_by_token(token)
        
        if not invitation or invitation.status != 'pending':
            raise ValueError('Invalid or expired invitation')
        
        if invitation.is_expired():
            invitation.status = 'expired'
            await self.save_invitation(invitation)
            raise ValueError('Invitation has expired')
        
        # Add user to team
        await self.add_user_to_team(user_id, invitation.team_id, invitation.role)
        
        # Update invitation status
        invitation.status = 'accepted'
        invitation.accepted_at = datetime.utcnow()
        invitation.user_id = user_id
        
        await self.save_invitation(invitation)
        
        # Send welcome notification
        await self.send_welcome_notification(user_id, invitation.team_id)
        
        return invitation

Bulk Invitation Implementation

# Bulk invitation service
class BulkInvitationService:
    def __init__(self):
        self.batch_size = 100
        self.max_concurrent_emails = 10
    
    async def create_bulk_invitations(self, invitations_data):
        """Create multiple invitations"""
        results = []
        
        for batch in self.chunk_list(invitations_data, self.batch_size):
            batch_results = await self.create_invitation_batch(batch)
            results.extend(batch_results)
        
        return results
    
    async def create_invitation_batch(self, batch):
        """Create a batch of invitations"""
        invitations = []
        
        for invitation_data in batch:
            try:
                invitation = await self.create_single_invitation(invitation_data)
                invitations.append({
                    'email': invitation_data['email'],
                    'status': 'created',
                    'invitation_id': invitation.id,
                    'error': None
                })
            except Exception as e:
                invitations.append({
                    'email': invitation_data['email'],
                    'status': 'error',
                    'invitation_id': None,
                    'error': str(e)
                })
        
        # Send emails concurrently
        await self.send_bulk_emails(invitations)
        
        return invitations
    
    async def send_bulk_emails(self, invitations):
        """Send bulk invitation emails"""
        valid_invitations = [inv for inv in invitations if inv['status'] == 'created']
        
        semaphore = asyncio.Semaphore(self.max_concurrent_emails)
        
        async def send_email(invitation):
            async with semaphore:
                try:
                    await self.send_invitation_email(invitation['invitation_id'])
                    return {'email': invitation['email'], 'status': 'sent'}
                except Exception as e:
                    return {'email': invitation['email'], 'status': 'error', 'error': str(e)}
        
        tasks = [send_email(inv) for inv in valid_invitations]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return results