143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
"""Domain models."""
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from sqlalchemy import String, Boolean, DateTime, ForeignKey, Text, Enum as SQLEnum
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship, backref
|
|
|
|
from app.database import Base
|
|
|
|
|
|
class DomainStatus(str, Enum):
|
|
"""Domain availability status."""
|
|
AVAILABLE = "available"
|
|
TAKEN = "taken"
|
|
DROPPING_SOON = "dropping_soon" # In transition/pending delete
|
|
ERROR = "error"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class Domain(Base):
|
|
"""Domain model for tracking domain names."""
|
|
|
|
__tablename__ = "domains"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, index=True)
|
|
name: Mapped[str] = mapped_column(String(255), index=True, nullable=False)
|
|
|
|
# Current status
|
|
status: Mapped[DomainStatus] = mapped_column(
|
|
SQLEnum(DomainStatus), default=DomainStatus.UNKNOWN
|
|
)
|
|
is_available: Mapped[bool] = mapped_column(Boolean, default=False)
|
|
|
|
# WHOIS data (optional)
|
|
registrar: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
|
expiration_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
|
deletion_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) # When domain will be fully deleted
|
|
|
|
# User relationship
|
|
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
|
|
user: Mapped["User"] = relationship("User", back_populates="domains")
|
|
|
|
# Timestamps
|
|
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
last_checked: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
|
# How the current status was derived (rdap_iana, whois, dns, etc.)
|
|
last_check_method: Mapped[str | None] = mapped_column(String(30), nullable=True)
|
|
|
|
# Check history relationship
|
|
checks: Mapped[list["DomainCheck"]] = relationship(
|
|
"DomainCheck", back_populates="domain", cascade="all, delete-orphan"
|
|
)
|
|
|
|
# Notification settings
|
|
notify_on_available: Mapped[bool] = mapped_column(Boolean, default=True)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Domain {self.name} ({self.status})>"
|
|
|
|
# ------------------------------------------------------------------
|
|
# Canonical status fields (API stability for Terminal consistency)
|
|
# ------------------------------------------------------------------
|
|
@property
|
|
def status_checked_at(self) -> datetime | None:
|
|
return self.last_checked
|
|
|
|
@property
|
|
def status_source(self) -> str | None:
|
|
return self.last_check_method
|
|
|
|
|
|
class DomainCheck(Base):
|
|
"""History of domain availability checks."""
|
|
|
|
__tablename__ = "domain_checks"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, index=True)
|
|
domain_id: Mapped[int] = mapped_column(ForeignKey("domains.id"), nullable=False)
|
|
|
|
# Check results
|
|
status: Mapped[DomainStatus] = mapped_column(SQLEnum(DomainStatus))
|
|
is_available: Mapped[bool] = mapped_column(Boolean)
|
|
|
|
# Details
|
|
response_data: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
|
|
# Timestamp
|
|
checked_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationship
|
|
domain: Mapped["Domain"] = relationship("Domain", back_populates="checks")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<DomainCheck {self.domain_id} at {self.checked_at}>"
|
|
|
|
|
|
class HealthStatus(str, Enum):
|
|
"""Domain health status levels."""
|
|
HEALTHY = "healthy"
|
|
WEAKENING = "weakening"
|
|
PARKED = "parked"
|
|
CRITICAL = "critical"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class DomainHealthCache(Base):
|
|
"""
|
|
Cached health check results for domains.
|
|
|
|
Updated daily by the scheduler to provide instant health status
|
|
without needing manual checks.
|
|
"""
|
|
|
|
__tablename__ = "domain_health_cache"
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True, index=True)
|
|
domain_id: Mapped[int] = mapped_column(ForeignKey("domains.id"), unique=True, nullable=False)
|
|
|
|
# Health status
|
|
status: Mapped[str] = mapped_column(String(20), default="unknown")
|
|
score: Mapped[int] = mapped_column(default=0)
|
|
|
|
# Signals (JSON array as text)
|
|
signals: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
|
|
# Layer data (JSON as text for flexibility)
|
|
dns_data: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
http_data: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
ssl_data: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
|
|
# Timestamp
|
|
checked_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationship - cascade delete when domain is deleted
|
|
domain: Mapped["Domain"] = relationship(
|
|
"Domain",
|
|
backref=backref("health_cache", cascade="all, delete-orphan", uselist=False)
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<DomainHealthCache {self.domain_id} status={self.status}>"
|
|
|