Spaces:
Sleeping
Sleeping
| """ | |
| INCIDENT Models - Support Requests | |
| Incidents are support requests from customers with existing services. | |
| Each incident can generate a support ticket for resolution. | |
| Workflow: | |
| 1. Customer reports problem with existing service | |
| 2. Support creates Incident (status='open') | |
| 3. Manager creates Ticket from Incident (source='incident') | |
| 4. Incident status changes to 'ticket_created' | |
| 5. When ticket completes, Incident status changes to 'resolved' | |
| """ | |
| from sqlalchemy import Column, String, Boolean, Text, DateTime, ForeignKey, CheckConstraint | |
| from sqlalchemy.dialects.postgresql import UUID, JSONB | |
| from sqlalchemy.orm import relationship | |
| from datetime import datetime | |
| from typing import Optional | |
| import uuid | |
| from app.models.base import BaseModel | |
| from app.models.enums import TicketPriority | |
| class Incident(BaseModel): | |
| """ | |
| Incident (Support Request) Model | |
| Represents support requests from customers with existing subscriptions. | |
| Incidents can be promoted to tickets for field technician dispatch. | |
| """ | |
| __tablename__ = "incidents" | |
| # Customer & Subscription Links | |
| customer_id = Column(UUID(as_uuid=True), ForeignKey("customers.id", ondelete="RESTRICT"), nullable=False, index=True) | |
| subscription_id = Column(UUID(as_uuid=True), ForeignKey("subscriptions.id", ondelete="SET NULL"), nullable=True) | |
| project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="RESTRICT"), nullable=False, index=True) | |
| # Incident Details | |
| incident_type = Column(Text, nullable=False) # 'no_service', 'slow_speed', 'equipment_fault', 'billing_issue' | |
| issue_description = Column(Text, nullable=False) | |
| priority = Column(String(20), nullable=False, default=TicketPriority.NORMAL.value) | |
| # Region Reference (inherited from Subscription for quick queries) | |
| project_region_id = Column(UUID(as_uuid=True), ForeignKey("project_regions.id", ondelete="SET NULL"), nullable=True) | |
| # Resolution Tracking | |
| # Status workflow: open → ticket_created → resolved → closed OR cancelled | |
| status = Column(Text, nullable=False, default='open') # 'open', 'ticket_created', 'resolved', 'closed', 'cancelled' | |
| is_ticket_created = Column(Boolean, nullable=False, default=False) | |
| resolved_at = Column(DateTime(timezone=True), nullable=True) | |
| resolution_description = Column(Text, nullable=True) | |
| cancellation_reason = Column(Text, nullable=True) | |
| # Metadata | |
| additional_metadata = Column(JSONB, nullable=False, default=dict, server_default='{}') | |
| # Relationships | |
| customer = relationship("Customer", back_populates="incidents") | |
| # subscription = relationship("Subscription", back_populates="incidents") # Add to Subscription model later | |
| project = relationship("Project", foreign_keys=[project_id]) | |
| project_region = relationship("ProjectRegion", foreign_keys=[project_region_id]) | |
| # ============================================ | |
| # COMPUTED PROPERTIES | |
| # ============================================ | |
| def is_open(self) -> bool: | |
| """Check if incident is open (awaiting ticket creation)""" | |
| return self.status == 'open' | |
| def is_resolved(self) -> bool: | |
| """Check if incident is resolved""" | |
| return self.status == 'resolved' | |
| def is_closed(self) -> bool: | |
| """Check if incident is closed""" | |
| return self.status == 'closed' | |
| def can_promote_to_ticket(self) -> bool: | |
| """Check if incident can be promoted to ticket""" | |
| return self.status == 'open' and not self.is_ticket_created | |
| # ============================================ | |
| # BUSINESS METHODS | |
| # ============================================ | |
| def mark_ticket_created(self): | |
| """Mark that ticket has been created from this incident""" | |
| self.status = 'ticket_created' | |
| self.is_ticket_created = True | |
| def mark_resolved(self, resolution_description: Optional[str] = None): | |
| """Mark incident as resolved""" | |
| self.status = 'resolved' | |
| self.resolved_at = datetime.utcnow() | |
| if resolution_description: | |
| self.resolution_description = resolution_description | |
| def mark_closed(self): | |
| """Mark incident as closed (customer confirmed resolution)""" | |
| self.status = 'closed' | |
| def cancel(self, reason: str): | |
| """Cancel incident (false alarm or customer withdrew)""" | |
| self.status = 'cancelled' | |
| self.cancellation_reason = reason | |
| def to_dict(self): | |
| """Convert to dictionary for serialization""" | |
| return { | |
| 'id': str(self.id), | |
| 'customer_id': str(self.customer_id), | |
| 'subscription_id': str(self.subscription_id) if self.subscription_id else None, | |
| 'project_id': str(self.project_id), | |
| 'incident_type': self.incident_type, | |
| 'issue_description': self.issue_description, | |
| 'priority': self.priority, | |
| 'project_region_id': str(self.project_region_id) if self.project_region_id else None, | |
| 'status': self.status, | |
| 'is_ticket_created': self.is_ticket_created, | |
| 'resolved_at': self.resolved_at.isoformat() if self.resolved_at else None, | |
| 'resolution_description': self.resolution_description, | |
| 'cancellation_reason': self.cancellation_reason, | |
| 'additional_metadata': self.additional_metadata, | |
| 'created_at': self.created_at.isoformat(), | |
| 'updated_at': self.updated_at.isoformat(), | |
| } | |