k.eaven
bec06bd909
- Added multi-guild support with comma-separated AllowedGuildIDs config - Per-guild embed tracking with isolated state - Per-server action locking to prevent race conditions across guilds - Cross-guild embed updates after power actions - Debug logging for embed update triggers - Fixed resource usage field formatting (brackets now properly closed) - Updated config.ini.sample, generate_config.py for backward compatibility - Updated README.md with multi-guild documentation - Updated tests for multi-guild support
118 lines
2.6 KiB
Python
118 lines
2.6 KiB
Python
"""
|
|
Data Models Module
|
|
|
|
Pydantic models for type-safe data handling throughout the application.
|
|
Provides validation and serialization for API responses and internal data.
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class ResourceAttributes(BaseModel):
|
|
"""Current resource usage attributes from panel API."""
|
|
|
|
current_state: str
|
|
resources: "ResourceUsage"
|
|
|
|
|
|
class ResourceUsage(BaseModel):
|
|
"""Individual resource metrics."""
|
|
|
|
cpu_absolute: float = 0.0
|
|
memory_bytes: int = 0
|
|
disk_bytes: int = 0
|
|
network_rx_bytes: int = 0
|
|
network_tx_bytes: int = 0
|
|
uptime: int = 0
|
|
|
|
|
|
class ServerAttributes(BaseModel):
|
|
"""Server attributes from panel API."""
|
|
|
|
identifier: str
|
|
name: str
|
|
description: Optional[str] = None
|
|
suspended: bool = False
|
|
limits: "ServerLimits" = Field(default_factory=lambda: ServerLimits())
|
|
|
|
|
|
class ServerLimits(BaseModel):
|
|
"""Server resource limits."""
|
|
|
|
cpu: int = 0
|
|
memory: int = 0
|
|
disk: int = 0
|
|
|
|
|
|
class ServerRelationships(BaseModel):
|
|
"""Server relationships data."""
|
|
|
|
allocations: Optional["AllocationsData"] = None
|
|
|
|
|
|
class AllocationsData(BaseModel):
|
|
"""Allocations data container."""
|
|
|
|
data: List["Allocation"] = Field(default_factory=list)
|
|
|
|
|
|
class Allocation(BaseModel):
|
|
"""Individual allocation (IP/port)."""
|
|
|
|
attributes: "AllocationAttributes"
|
|
|
|
|
|
class AllocationAttributes(BaseModel):
|
|
"""Allocation attributes."""
|
|
|
|
ip_alias: str = "Unknown"
|
|
port: int = 0
|
|
is_default: bool = False
|
|
|
|
|
|
class ServerData(BaseModel):
|
|
"""Complete server data from panel API."""
|
|
|
|
attributes: ServerAttributes
|
|
relationships: Optional[ServerRelationships] = None
|
|
|
|
|
|
class ServerResources(BaseModel):
|
|
"""Server resources response from panel API."""
|
|
|
|
attributes: ResourceAttributes
|
|
|
|
|
|
class APIResponse(BaseModel):
|
|
"""Generic API response wrapper."""
|
|
|
|
status: str = "success"
|
|
data: Any = None
|
|
message: Optional[str] = None
|
|
errors: Optional[List[Dict[str, Any]]] = None
|
|
|
|
|
|
class ServerListResponse(BaseModel):
|
|
"""Server list response from panel API."""
|
|
|
|
data: List[Dict[str, Any]] = Field(default_factory=list)
|
|
meta: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class EmbedLocation(BaseModel):
|
|
"""Embed location tracking data with guild context."""
|
|
|
|
guild_id: int
|
|
channel_id: str
|
|
message_id: str
|
|
|
|
# Composite key for storage
|
|
@property
|
|
def storage_key(self) -> str:
|
|
return f"{self.guild_id}:{self.channel_id}:{self.message_id}"
|
|
|
|
|
|
# Rebuild models to resolve forward references
|
|
ResourceAttributes.model_rebuild()
|
|
ServerAttributes.model_rebuild() |