AWS-Panel/backend/modules/instances/schemas.py

from datetime import datetime
from typing import Any, Dict, List, Optional
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

from backend.modules.instances.models import InstanceDesiredStatus, InstanceStatus


class InstanceFilterParams(BaseModel):
    credential_id: Optional[int] = None
    account_id: Optional[str] = None
    region: Optional[str] = None
    status: Optional[InstanceStatus] = None
    customer_id: Optional[int] = None
    keyword: Optional[str] = Field(default=None, description="search name/id/ip")
    offset: int = 0
    limit: int = 20


class InstanceCreateRequest(BaseModel):
    credential_id: int
    account_id: str
    region: str
    ami_id: Optional[str] = None
    os_name: Optional[str] = None
    instance_type: str
    key_name: Optional[str] = None
    subnet_id: Optional[str] = None
    security_group_ids: List[str] = Field(default_factory=list)
    name_tag: Optional[str] = None
    customer_id: Optional[int] = None
    os_family: str = "ubuntu-22.04"
    login_username: Optional[str] = None
    login_password: Optional[str] = None
    user_data_mode: str = Field(default="auto_root", description="auto_root|custom|none")
    custom_user_data: Optional[str] = None
    volume_size: int = Field(default=20, description="GiB")
    volume_type: str = Field(default="gp3")
    count: int = Field(default=1, ge=1, le=100, description="Number of instances to create")


class InstanceOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    customer_id: int
    credential_id: Optional[int]
    account_id: str
    region: str
    az: Optional[str] = None
    credential_name: Optional[str] = None
    credential_label: Optional[str] = None
    owner_name: Optional[str] = None
    customer_name: Optional[str] = None
    instance_id: str
    name_tag: Optional[str] = None
    instance_type: str
    instance_vcpus: Optional[int] = None
    instance_memory_gib: Optional[float] = None
    instance_network_perf: Optional[str] = None
    ami_id: Optional[str] = None
    os_name: Optional[str] = None
    os_pretty_name: Optional[str] = None
    key_name: Optional[str] = None
    public_ip: Optional[str] = None
    private_ip: Optional[str] = None
    status: InstanceStatus
    desired_status: Optional[InstanceDesiredStatus] = None
    security_groups: Optional[List[str]] = None
    subnet_id: Optional[str] = None
    vpc_id: Optional[str] = None
    launched_at: Optional[datetime] = None
    terminated_at: Optional[datetime] = None
    last_sync: Optional[datetime] = None
    last_cloud_state: Optional[Any] = None
    created_at: datetime
    updated_at: datetime


class InstanceListResponse(BaseModel):
    items: List[InstanceOut]
    total: int


class InstanceSyncRequest(BaseModel):
    credential_id: Optional[int] = None
    region: Optional[str] = None
    customer_id: Optional[int] = None


class InstanceCreateResponse(InstanceOut):
    login_username: Optional[str] = None
    login_password: Optional[str] = None


class BatchInstancesActionIn(BaseModel):
    instance_ids: List[int]
    action: Literal["start", "stop", "reboot", "terminate", "sync"]


class BatchInstancesActionOut(BaseModel):
    action: str
    requested: List[int]
    accepted: List[int]
    skipped: List[int]
    errors: Dict[str, str] = Field(default_factory=dict)


class BatchInstancesByIpIn(BaseModel):
    ips: List[str]
    action: Literal["start", "stop", "reboot", "terminate", "sync"]
    credential_id: Optional[int] = None
    region: Optional[str] = None


class BatchInstancesByIpOut(BaseModel):
    ips_requested: List[str]
    ips_matched: List[str]
    ips_unmatched: List[str]
    result: BatchInstancesActionOut


class InstanceIdsExportIn(BaseModel):
    instance_ids: List[int]