from __future__ import annotations from datetime import datetime from enum import Enum from typing import Optional from sqlalchemy import DateTime, Enum as SAEnum, ForeignKey, Index, JSON, String, UniqueConstraint, text from sqlalchemy.dialects.mysql import BIGINT from sqlalchemy.orm import Mapped, mapped_column, relationship from backend.db.base import Base class InstanceStatus(str, Enum): PENDING = "PENDING" RUNNING = "RUNNING" STOPPING = "STOPPING" STOPPED = "STOPPED" SHUTTING_DOWN = "SHUTTING_DOWN" TERMINATED = "TERMINATED" UNKNOWN = "UNKNOWN" class InstanceDesiredStatus(str, Enum): RUNNING = "RUNNING" STOPPED = "STOPPED" TERMINATED = "TERMINATED" class Instance(Base): __tablename__ = "instances" __table_args__ = ( UniqueConstraint("account_id", "region", "instance_id", name="uniq_instance_cloud"), Index("idx_instances_customer", "customer_id"), Index("idx_instances_status", "status"), Index("idx_instances_region", "region"), Index("idx_instances_last_sync", "last_sync"), ) id: Mapped[int] = mapped_column(BIGINT(unsigned=True), primary_key=True, autoincrement=True) customer_id: Mapped[int] = mapped_column( ForeignKey("customers.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=False ) credential_id: Mapped[Optional[int]] = mapped_column( ForeignKey("aws_credentials.id", ondelete="SET NULL", onupdate="CASCADE") ) account_id: Mapped[str] = mapped_column(String(32), nullable=False) region: Mapped[str] = mapped_column(String(32), nullable=False) az: Mapped[Optional[str]] = mapped_column(String(32)) instance_id: Mapped[str] = mapped_column(String(32), nullable=False) name_tag: Mapped[Optional[str]] = mapped_column(String(255)) instance_type: Mapped[str] = mapped_column(String(64), nullable=False) ami_id: Mapped[Optional[str]] = mapped_column(String(64)) os_name: Mapped[Optional[str]] = mapped_column(String(128)) key_name: Mapped[Optional[str]] = mapped_column(String(128)) public_ip: Mapped[Optional[str]] = mapped_column(String(45)) private_ip: Mapped[Optional[str]] = mapped_column(String(45)) status: Mapped[InstanceStatus] = mapped_column( SAEnum(InstanceStatus), nullable=False, server_default=text("'UNKNOWN'") ) desired_status: Mapped[Optional[InstanceDesiredStatus]] = mapped_column(SAEnum(InstanceDesiredStatus)) security_groups: Mapped[Optional[dict]] = mapped_column(JSON) subnet_id: Mapped[Optional[str]] = mapped_column(String(64)) vpc_id: Mapped[Optional[str]] = mapped_column(String(64)) launched_at: Mapped[Optional[datetime]] = mapped_column(DateTime) terminated_at: Mapped[Optional[datetime]] = mapped_column(DateTime) last_sync: Mapped[Optional[datetime]] = mapped_column(DateTime) last_cloud_state: Mapped[Optional[dict]] = mapped_column(JSON) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"), nullable=False) updated_at: Mapped[datetime] = mapped_column( DateTime, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP"), nullable=False ) customer: Mapped["Customer"] = relationship("Customer") credential: Mapped[Optional["AWSCredential"]] = relationship("AWSCredential", back_populates="instances") job_items: Mapped[list["JobItem"]] = relationship("JobItem", back_populates="instance") @property def credential_name(self) -> Optional[str]: if getattr(self, "credential", None): return self.credential.name return None @property def credential_label(self) -> Optional[str]: if getattr(self, "credential", None): return f"{self.credential.name} ({self.credential.account_id})" return None @property def owner_name(self) -> Optional[str]: if getattr(self, "customer", None): return self.customer.name or self.customer.contact_email return None @property def customer_name(self) -> Optional[str]: return self.owner_name