# services/auth/app/models/onboarding.py """ User onboarding progress models """ from sqlalchemy import Column, String, Boolean, DateTime, Text, ForeignKey, JSON, UniqueConstraint from sqlalchemy.dialects.postgresql import UUID from datetime import datetime, timezone import uuid from shared.database.base import Base class UserOnboardingProgress(Base): """User onboarding progress tracking model""" __tablename__ = "user_onboarding_progress" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) # Step tracking step_name = Column(String(50), nullable=False) completed = Column(Boolean, default=False, nullable=False) completed_at = Column(DateTime(timezone=True)) # Additional step data (JSON field for flexibility) step_data = Column(JSON, default=dict) # Timestamps created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) # Unique constraint to prevent duplicate step entries per user __table_args__ = ( UniqueConstraint('user_id', 'step_name', name='uq_user_step'), {'extend_existing': True} ) def __repr__(self): return f"" def to_dict(self): """Convert to dictionary""" return { "id": str(self.id), "user_id": str(self.user_id), "step_name": self.step_name, "completed": self.completed, "completed_at": self.completed_at.isoformat() if self.completed_at else None, "step_data": self.step_data or {}, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None } class UserOnboardingSummary(Base): """User onboarding summary for quick lookups""" __tablename__ = "user_onboarding_summary" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, unique=True, index=True) # Summary fields current_step = Column(String(50), nullable=False, default="user_registered") next_step = Column(String(50)) completion_percentage = Column(String(10), default="0.0") # Store as string for precision fully_completed = Column(Boolean, default=False) # Progress tracking steps_completed_count = Column(String(10), default="0") # Store as string: "3/5" # Timestamps created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) last_activity_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) def __repr__(self): return f"" def to_dict(self): """Convert to dictionary""" return { "id": str(self.id), "user_id": str(self.user_id), "current_step": self.current_step, "next_step": self.next_step, "completion_percentage": float(self.completion_percentage) if self.completion_percentage else 0.0, "fully_completed": self.fully_completed, "steps_completed_count": self.steps_completed_count, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "last_activity_at": self.last_activity_at.isoformat() if self.last_activity_at else None }