""" Core graph models for LabGraph. Graph structure: Node — vertices (any infrastructure component) Edge — directed relationships between nodes Network — IP network segments (VLANs/subnets) WikiPage — Markdown documentation anchored to a node HeartbeatLog — time-series liveness records per node """ import uuid import markdown as md from django.db import models class Node(models.Model): """ Polymorphic infrastructure node — the vertex type in the LabGraph. node_type drives rendering and determines what metadata keys are expected in the JSONField. Pydantic schemas in schemas.py validate metadata content at the API boundary before writes reach the ORM. """ class NodeType(models.TextChoices): LOCATION = "location", "Physical Location" HARDWARE = "hardware", "Physical Hardware" HYPERVISOR = "hypervisor", "Hypervisor Host" VM = "vm", "Virtual Machine" CONTAINER = "container", "Container" APPLICATION = "application", "Application / Service" NETWORK_DEVICE = "network_device", "Network Device" class Status(models.TextChoices): ONLINE = "online", "Online" OFFLINE = "offline", "Offline" DEGRADED = "degraded", "Degraded" UNKNOWN = "unknown", "Unknown" MAINTENANCE = "maintenance", "Maintenance" class DiscoverySource(models.TextChoices): MANUAL = "manual", "Manually Added" PROXMOX = "proxmox", "Proxmox API" NMAP = "nmap", "Nmap Scan" SNMP = "snmp", "SNMP" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) label = models.CharField(max_length=255, db_index=True) node_type = models.CharField( max_length=30, choices=NodeType.choices, db_index=True ) status = models.CharField( max_length=20, choices=Status.choices, default=Status.UNKNOWN, db_index=True, ) ip_address = models.GenericIPAddressField(null=True, blank=True, db_index=True) # Stored as colon-separated hex: "aa:bb:cc:dd:ee:ff" mac_address = models.CharField(max_length=17, null=True, blank=True) # Rated power draw in watts — used by the power cost estimator (Phase 3) wattage = models.DecimalField(max_digits=8, decimal_places=2, null=True, blank=True) # Flexible per-type data; validated by Pydantic at the API boundary metadata = models.JSONField(default=dict, blank=True) discovery_source = models.CharField( max_length=20, choices=DiscoverySource.choices, default=DiscoverySource.MANUAL, ) # ID in the originating external system, e.g. Proxmox VMID or Docker container ID external_id = models.CharField(max_length=255, null=True, blank=True, db_index=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["label"] indexes = [ models.Index(fields=["node_type", "status"]), models.Index(fields=["discovery_source", "external_id"]), ] verbose_name = "Node" verbose_name_plural = "Nodes" def __str__(self) -> str: return f"{self.label} ({self.get_node_type_display()})" class Edge(models.Model): """ Directed edge between two Nodes in the infrastructure graph. Direction: source → target. Multiple edge types allow different relationship semantics to coexist (e.g. a VM has a parent_child edge to its hypervisor AND a network edge to a switch). """ class EdgeType(models.TextChoices): PARENT_CHILD = "parent_child", "Parent / Child" NETWORK = "network", "Network Connection" DEPENDENCY = "dependency", "Software Dependency" PHYSICAL = "physical", "Physical Connection" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) source = models.ForeignKey( Node, on_delete=models.CASCADE, related_name="outgoing_edges" ) target = models.ForeignKey( Node, on_delete=models.CASCADE, related_name="incoming_edges" ) edge_type = models.CharField( max_length=20, choices=EdgeType.choices, db_index=True ) weight = models.FloatField(default=1.0) label = models.CharField(max_length=255, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = [("source", "target", "edge_type")] indexes = [models.Index(fields=["edge_type"])] verbose_name = "Edge" verbose_name_plural = "Edges" def __str__(self) -> str: return f"{self.source.label} → {self.target.label} [{self.edge_type}]" class Network(models.Model): """ An IP network segment that nodes attach to. Nodes reference Networks via metadata or a future ManyToMany through-model (Phase 2 IPAM). Networks are separate entities from Nodes so that VLAN and subnet data can be managed independently. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) name = models.CharField(max_length=255) vlan_id = models.PositiveSmallIntegerField( null=True, blank=True, help_text="802.1Q VLAN tag (1–4094)", ) # CIDR notation, e.g. "192.168.10.0/24" cidr = models.CharField(max_length=43, db_index=True) gateway = models.GenericIPAddressField(null=True, blank=True) description = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["cidr"] verbose_name = "Network" verbose_name_plural = "Networks" def __str__(self) -> str: vlan_str = f" (VLAN {self.vlan_id})" if self.vlan_id else "" return f"{self.name} {self.cidr}{vlan_str}" class WikiPage(models.Model): """ Human-authored documentation anchored to a single Node. Content is stored as raw Markdown. rendered_html() converts it to safe HTML at render time using the Markdown package. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) node = models.OneToOneField( Node, on_delete=models.CASCADE, related_name="wiki_page" ) content = models.TextField(blank=True, help_text="Markdown content") last_edited_by = models.ForeignKey( "auth.User", null=True, blank=True, on_delete=models.SET_NULL, related_name="wiki_edits", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = "Wiki Page" verbose_name_plural = "Wiki Pages" def __str__(self) -> str: return f"Wiki: {self.node.label}" def rendered_html(self) -> str: """Convert Markdown content to safe HTML.""" return md.markdown( self.content, extensions=["fenced_code", "tables", "nl2br"], ) class HeartbeatLog(models.Model): """ Time-series liveness record for a Node. Written by tasks/heartbeat.py on every check cycle. High-volume — records accumulate fast. The maintenance task prunes rows older than 30 days nightly. """ class CheckType(models.TextChoices): ICMP = "icmp", "ICMP Ping" TCP = "tcp", "TCP Port Check" HTTP = "http", "HTTP/S Check" SNMP = "snmp", "SNMP Poll" class HeartbeatStatus(models.TextChoices): UP = "up", "Up" DOWN = "down", "Down" TIMEOUT = "timeout", "Timeout" ERROR = "error", "Error" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) node = models.ForeignKey( Node, on_delete=models.CASCADE, related_name="heartbeat_logs" ) timestamp = models.DateTimeField(auto_now_add=True, db_index=True) status = models.CharField( max_length=10, choices=HeartbeatStatus.choices, db_index=True ) check_type = models.CharField(max_length=10, choices=CheckType.choices) # Null on timeout or error — response never completed response_time_ms = models.PositiveIntegerField(null=True, blank=True) error_message = models.TextField(blank=True) class Meta: ordering = ["-timestamp"] indexes = [ models.Index(fields=["node", "-timestamp"]), models.Index(fields=["status", "-timestamp"]), ] verbose_name = "Heartbeat Log" verbose_name_plural = "Heartbeat Logs" def __str__(self) -> str: return f"{self.node.label} [{self.status}] @ {self.timestamp:%Y-%m-%d %H:%M:%S}"