Spaces:
Sleeping
Sleeping
| """ | |
| User Invitation Model | |
| """ | |
| from sqlalchemy import Column, String, Boolean, DateTime, Text, ForeignKey | |
| from sqlalchemy.dialects.postgresql import UUID, JSONB | |
| from app.models.base import BaseModel | |
| class UserInvitation(BaseModel): | |
| """ | |
| User Invitation model - Tracks invitations before Supabase Auth account creation | |
| Users are created in Supabase Auth ONLY after accepting invitation. | |
| This table manages the invitation lifecycle: pending → accepted/expired/cancelled | |
| """ | |
| __tablename__ = "user_invitations" | |
| # Invitation Details | |
| email = Column(String(255), nullable=False) | |
| phone = Column(String(50), nullable=True) | |
| invited_name = Column(String(200), nullable=True) # Optional full name of invitee | |
| invited_role = Column(String(50), nullable=False) # app_role ENUM | |
| # Organization Links | |
| client_id = Column(UUID(as_uuid=True), ForeignKey('clients.id', ondelete='SET NULL'), nullable=True) | |
| contractor_id = Column(UUID(as_uuid=True), ForeignKey('contractors.id', ondelete='SET NULL'), nullable=True) | |
| # Project Context (for project-specific invitations) | |
| project_id = Column(UUID(as_uuid=True), ForeignKey('projects.id', ondelete='CASCADE'), nullable=True) | |
| project_role_id = Column(UUID(as_uuid=True), ForeignKey('project_roles.id', ondelete='SET NULL'), nullable=True) | |
| project_region_id = Column(UUID(as_uuid=True), ForeignKey('project_regions.id', ondelete='SET NULL'), nullable=True) | |
| project_subcontractor_id = Column(UUID(as_uuid=True), ForeignKey('project_subcontractors.id', ondelete='SET NULL'), nullable=True) | |
| # Token & Status | |
| token = Column(Text, nullable=False, unique=True) | |
| status = Column(String(50), default='pending') # invitation_status ENUM | |
| invitation_method = Column(String(50), default='whatsapp') # invitation_method ENUM | |
| # Lifecycle | |
| invited_by_user_id = Column(UUID(as_uuid=True), ForeignKey('users.id', ondelete='CASCADE'), nullable=False) | |
| invited_at = Column(DateTime(timezone=True), nullable=False) | |
| expires_at = Column(DateTime(timezone=True), nullable=False) | |
| accepted_at = Column(DateTime(timezone=True), nullable=True) | |
| # Delivery Tracking | |
| whatsapp_sent = Column(Boolean, default=False) | |
| whatsapp_sent_at = Column(DateTime(timezone=True), nullable=True) | |
| whatsapp_error = Column(Text, nullable=True) | |
| email_sent = Column(Boolean, default=False) | |
| email_sent_at = Column(DateTime(timezone=True), nullable=True) | |
| email_error = Column(Text, nullable=True) | |
| # Metadata | |
| invitation_metadata = Column(JSONB, default={}) | |
| def __repr__(self): | |
| return f"<UserInvitation(email='{self.email}', role='{self.invited_role}', status='{self.status}')>" | |
| def is_expired(self) -> bool: | |
| """Check if invitation has expired""" | |
| from datetime import datetime, timezone | |
| return self.expires_at < datetime.now(timezone.utc) | |
| def is_pending(self) -> bool: | |
| """Check if invitation is still pending""" | |
| return self.status == 'pending' and not self.is_expired | |
| def organization_id(self): | |
| """Get the organization ID (client or contractor)""" | |
| return self.client_id or self.contractor_id | |
| def organization_type(self) -> str: | |
| """Get organization type""" | |
| if self.client_id: | |
| return 'client' | |
| elif self.contractor_id: | |
| return 'contractor' | |
| return 'platform' | |
| def is_project_invitation(self) -> bool: | |
| """Check if this is a project-specific invitation""" | |
| return self.project_id is not None | |