import uuid from datetime import datetime, timezone from typing import Optional, Any, TYPE_CHECKING from sqlalchemy import String, Text, DateTime, ForeignKey, Boolean, Integer, Index, CheckConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.dialects.postgresql import UUID, JSONB from app.core.database import Base if TYPE_CHECKING: from app.models.user import User from app.models.team import Team from app.models.account import Account from app.models.session import Session from app.models.category import TreeCategory from app.models.tag import TreeTag from app.models.folder import UserFolder from app.models.tree_share import TreeShare class Tree(Base): __tablename__ = "trees" __table_args__ = ( CheckConstraint( "visibility IN ('private', 'team', 'link', 'public')", name='ck_trees_visibility' ), CheckConstraint( "status IN ('draft', 'published')", name='ck_trees_status' ), CheckConstraint( "tree_type IN ('troubleshooting', 'procedural')", name='ck_trees_tree_type' ), ) id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 ) name: Mapped[str] = mapped_column(String(255), nullable=False) description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Legacy category field - kept for backward compatibility # New code should use category_id instead category: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, index=True) # New category relationship category_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("tree_categories.id", ondelete="SET NULL"), nullable=True, index=True ) tree_type: Mapped[str] = mapped_column( String(20), nullable=False, default='troubleshooting', server_default='troubleshooting', index=True, comment="Tree type: troubleshooting (branching decision tree) or procedural (linear step-by-step flow)" ) tree_structure: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) intake_form: Mapped[Optional[list[dict[str, Any]]]] = mapped_column( JSONB, nullable=True, comment="Intake form field definitions for procedural flows (JSONB array)" ) author_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id"), nullable=True ) team_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("teams.id"), nullable=True, index=True ) account_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"), nullable=True, index=True ) is_active: Mapped[bool] = mapped_column(Boolean, default=True) is_public: Mapped[bool] = mapped_column(Boolean, default=False, index=True) is_default: Mapped[bool] = mapped_column(Boolean, default=False, index=True) visibility: Mapped[str] = mapped_column( String(20), nullable=False, default='team', index=True, comment="Visibility level: private (author only), team (account members), link (share token), public (all users)" ) status: Mapped[str] = mapped_column( String(20), nullable=False, default='published', index=True, comment="Status: draft (work in progress) or published (validated and available)" ) deleted_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True, index=True ) deleted_by: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id"), nullable=True ) version: Mapped[int] = mapped_column(Integer, default=1) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc) ) usage_count: Mapped[int] = mapped_column(Integer, default=0) # Fork tracking parent_tree_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("trees.id", ondelete="SET NULL"), nullable=True, index=True ) fork_reason: Mapped[Optional[str]] = mapped_column( String(255), nullable=True, comment="Brief reason: 'Added Cisco Meraki steps for our network'" ) parent_updated_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True, comment="Snapshot of parent's updated_at when fork created. Compare to detect parent updates." ) # Fork lineage tracking root_tree_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("trees.id", ondelete="SET NULL"), nullable=True, index=True, comment="Original tree at root of fork chain (NULL for non-forked trees)" ) fork_depth: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", comment="Fork depth: 0 = original, 1 = direct fork, 2 = fork of fork, etc." ) # Relationships author: Mapped[Optional["User"]] = relationship("User", foreign_keys=[author_id], back_populates="trees") team: Mapped[Optional["Team"]] = relationship("Team", back_populates="trees") account: Mapped[Optional["Account"]] = relationship("Account", foreign_keys=[account_id], back_populates="trees") # Fork relationships (self-referential) parent: Mapped[Optional["Tree"]] = relationship( "Tree", remote_side="Tree.id", foreign_keys=[parent_tree_id], back_populates="forks" ) forks: Mapped[list["Tree"]] = relationship( "Tree", foreign_keys=[parent_tree_id], back_populates="parent" ) root: Mapped[Optional["Tree"]] = relationship( "Tree", remote_side="Tree.id", foreign_keys=[root_tree_id] ) sessions: Mapped[list["Session"]] = relationship("Session", back_populates="tree") shares: Mapped[list["TreeShare"]] = relationship( "TreeShare", back_populates="tree", cascade="all, delete-orphan" ) # Organization relationships category_rel: Mapped[Optional["TreeCategory"]] = relationship("TreeCategory", back_populates="trees") tags: Mapped[list["TreeTag"]] = relationship( "TreeTag", secondary="tree_tag_assignments", back_populates="trees" ) folders: Mapped[list["UserFolder"]] = relationship( "UserFolder", secondary="user_folder_trees", back_populates="trees" ) # Full-text search index will be created in migration @property def tag_names(self) -> list[str]: """Returns list of tag names for this tree.""" return [tag.name for tag in self.tags] if self.tags else []