✨ feat: add Pydantic models for Deployment, Disk, ManagedService, Network, Organization, Package, Project, Secret, StaticRoute, and User

Author: guptadev21

rapyuta_io_sdk_v2/pydantic_models/__init__.py

@@ -0,0 +1,13 @@
+from .secret import Secret as Secret, SecretList as SecretList
+from .staticroute import StaticRoute as StaticRoute, StaticRouteList as StaticRouteList
+from .disk import Disk as Disk, DiskList as DiskList
+from .package import Package as Package, PackageList as PackageList
+from .deployment import Deployment as Deployment, DeploymentList as DeploymentList
+from .project import Project as Project, ProjectList as ProjectList
+from .network import Network as Network, NetworkList as NetworkList
+from .managedservice import (
+    ManagedService as ManagedService,
+    ManagedServiceList as ManagedServiceList,
+)
+from .user import User as User
+from .organization import Organization as Organization

rapyuta_io_sdk_v2/pydantic_models/deployment.py

@@ -0,0 +1,259 @@
+"""
+Pydantic models for Deployment resource validation.
+
+This module contains Pydantic models that correspond to the Deployment JSON schema,
+providing validation for Deployment resources to help users identify missing or
+incorrect fields.
+"""
+
+from typing import Literal
+from pydantic import BaseModel, Field, model_validator
+
+from rapyuta_io_sdk_v2.pydantic_models.utils import (
+    BaseMetadata,
+    BaseList,
+    Depends,
+    DeploymentPhase,
+    DeploymentStatusType,
+    ExecutableStatusType,
+    RestartPolicy,
+    Runtime,
+)
+
+
+class StringMap(dict[str, str]):
+    pass
+
+
+# --- Depends Models ---
+class PackageDepends(Depends):
+    kind: Literal["package"] = Field(default="package")
+
+
+class DeploymentMetadata(BaseMetadata):
+    """Metadata for Deployment resource."""
+
+    depends: PackageDepends | None = None
+
+
+class EnvArgsSpec(BaseModel):
+    name: str
+    value: str | None = None
+
+
+class DeploymentVolume(BaseModel):
+    """Unified volume spec matching Go DeploymentVolume struct."""
+
+    execName: str | None = None
+    mountPath: str | None = None
+    subPath: str | None = None
+    uid: int | None = None
+    gid: int | None = None
+    perm: int | None = None
+    depends: Depends | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_empty_depends(cls, data):
+        """Handle empty depends dictionaries by converting them to None."""
+        if isinstance(data, dict) and "depends" in data:
+            depends = data["depends"]
+            # If depends is an empty dictionary, set it to None
+            if isinstance(depends, dict) and not depends:
+                data["depends"] = None
+        return data
+
+
+class DeploymentStaticRoute(BaseModel):
+    """Static route configuration matching Go DeploymentStaticRoute struct."""
+
+    name: str | None = None
+    url: str | None = None
+    depends: Depends | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_empty_depends(cls, data):
+        """Handle empty depends dictionaries by converting them to None."""
+        if isinstance(data, dict) and "depends" in data:
+            depends = data["depends"]
+            # If depends is an empty dictionary, set it to None
+            if isinstance(depends, dict) and not depends:
+                data["depends"] = None
+        return data
+
+
+class ManagedServiceSpec(BaseModel):
+    depends: dict[str, str] | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_empty_depends(cls, data):
+        """Handle empty depends dictionaries by converting them to None."""
+        if isinstance(data, dict) and "depends" in data:
+            depends = data["depends"]
+            # If depends is an empty dictionary, set it to None
+            if isinstance(depends, dict) and not depends:
+                data["depends"] = None
+        return data
+
+
+class DeploymentROSNetwork(BaseModel):
+    """ROS Network configuration matching Go DeploymentROSNetwork struct."""
+
+    domainID: int = Field(description="ROS Domain ID")
+    depends: Depends | None = None
+    interface: str | None = Field(default=None, description="Network interface")
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_empty_depends(cls, data):
+        """Handle empty depends dictionaries by converting them to None."""
+        if isinstance(data, dict) and "depends" in data:
+            depends = data["depends"]
+            # If depends is an empty dictionary, set it to None
+            if isinstance(depends, dict) and not depends:
+                data["depends"] = None
+        return data
+
+
+class DeploymentParamConfig(BaseModel):
+    """Param configuration matching Go DeploymentParamConfig struct."""
+
+    enabled: bool | None = None
+    trees: list[str] | None = None
+    blockUntilSynced: bool | None = Field(default=False)
+
+
+class DeploymentVPNConfig(BaseModel):
+    """VPN configuration matching Go DeploymentVPNConfig struct."""
+
+    enabled: bool | None = Field(default=False)
+
+
+class DeploymentFeatures(BaseModel):
+    """Features configuration matching Go DeploymentFeatures struct."""
+
+    params: DeploymentParamConfig | None = None
+    vpn: DeploymentVPNConfig | None = None
+
+
+class DeploymentDevice(BaseModel):
+    """Device configuration matching Go DeploymentDevice struct."""
+
+    depends: Depends | None = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_empty_depends(cls, data):
+        """Handle empty depends dictionaries by converting them to None."""
+        if isinstance(data, dict) and "depends" in data:
+            depends = data["depends"]
+            # If depends is an empty dictionary, set it to None
+            if isinstance(depends, dict) and not depends:
+                data["depends"] = None
+        return data
+
+
+class DeploymentSpec(BaseModel):
+    runtime: Runtime
+    depends: list[Depends] | None = None
+    device: DeploymentDevice | None = None
+    restart: RestartPolicy = Field(default="always")
+    envArgs: list[EnvArgsSpec] | None = None
+    volumes: list[DeploymentVolume] | None = None
+    rosNetworks: list[DeploymentROSNetwork] | None = None
+    features: DeploymentFeatures | None = None
+    staticRoutes: list[DeploymentStaticRoute] | None = None
+    managedServices: list[ManagedServiceSpec] | None = None
+
+    @model_validator(mode="after")
+    def validate_runtime_and_volumes(self):
+        """Validate that runtime and volume configurations are compatible."""
+        if self.runtime == "device" and self.volumes:
+            # For device runtime, volumes should not have cloud-specific depends
+            for volume in self.volumes:
+                if volume.depends and hasattr(volume.depends, "kind"):
+                    # Device volumes should depend on disks, not cloud resources
+                    if volume.depends.kind in ["managedService", "cloudService"]:
+                        raise ValueError(
+                            f"Device runtime cannot use cloud volume dependency: {volume.depends.kind}"
+                        )
+        elif self.runtime == "cloud" and self.volumes:
+            # For cloud runtime, volumes should not have device-specific fields
+            for volume in self.volumes:
+                if any(
+                    [
+                        volume.uid is not None,
+                        volume.gid is not None,
+                        volume.perm is not None,
+                    ]
+                ):
+                    raise ValueError(
+                        "Cloud runtime cannot use device-specific volume fields: uid, gid, perm"
+                    )
+        return self
+
+
+class ExecutableStatus(BaseModel):
+    name: str | None = None
+    status: ExecutableStatusType | None = None
+    errorCode: str | None = Field(default=None, alias="error_code")
+    reason: str | None = None
+    restartCount: int | None = Field(default=None, alias="restart_count")  # int32 in Go
+    exitCode: int | None = Field(default=None, alias="exit_code")  # int32 in Go
+
+
+class DependentDeploymentStatus(BaseModel):
+    name: str | None = None
+    guid: str | None = None
+    status: DeploymentStatusType | None = None
+    phase: DeploymentPhase | None = None
+    errorCodes: list[str] | None = Field(default=None, alias="error_codes")
+
+
+class DependentNetworkStatus(BaseModel):
+    name: str | None = None
+    guid: str | None = None
+    status: DeploymentStatusType | None = None
+    phase: DeploymentPhase | None = None
+    errorCodes: list[str] | None = Field(default=None, alias="error_codes")
+
+
+class DependentDiskStatus(BaseModel):
+    name: str | None = None
+    guid: str | None = None
+    status: str | None = None
+    errorCode: str | None = Field(default=None, alias="error_codes")
+
+
+class Dependencies(BaseModel):
+    deployments: list[DependentDeploymentStatus] | None = None
+    networks: list[DependentNetworkStatus] | None = None
+    disks: list[DependentDiskStatus] | None = Field(default=None, alias="disk")
+
+
+class DeploymentStatus(BaseModel):
+    phase: DeploymentPhase | None = None
+    status: DeploymentStatusType | None = None
+    errorCodes: list[str] | None = Field(default=None, alias="error_codes")
+    executablesStatus: dict[str, ExecutableStatus] | None = Field(
+        default=None, alias="executables_status"
+    )
+    dependencies: Dependencies | None = None
+
+
+class Deployment(BaseModel):
+    """Deployment model."""
+
+    apiVersion: str | None = None
+    kind: str | None = None
+    metadata: DeploymentMetadata
+    spec: DeploymentSpec
+    status: DeploymentStatus | None = None
+
+
+class DeploymentList(BaseList[Deployment]):
+    """List of deployments using BaseList."""
+
+    pass

rapyuta_io_sdk_v2/pydantic_models/disk.py

@@ -0,0 +1,68 @@
+"""
+Pydantic models for Disk resource validation.
+
+This module contains Pydantic models that correspond to the Disk JSON schema,
+providing validation for Disk resources to help users identify missing or
+incorrect fields.
+"""
+
+from typing import Literal
+from pydantic import BaseModel, Field, field_validator
+
+from rapyuta_io_sdk_v2.pydantic_models.utils import BaseMetadata, BaseList, Runtime
+
+
+class DiskBound(BaseModel):
+    deployment_guid: str | None
+    deployment_name: str | None
+
+
+class DiskSpec(BaseModel):
+    """Specification for Disk resource."""
+
+    runtime: Runtime = Field(
+        default="cloud", description="Runtime environment for the disk"
+    )
+    capacity: int | float | None = Field(default=None, description="Disk capacity in GB")
+
+    @field_validator("capacity")
+    @classmethod
+    def validate_capacity(cls, v):
+        """Validate disk capacity against allowed values."""
+        if v is not None:
+            allowed_capacities = [4, 8, 16, 32, 64, 128, 256, 512]
+            if v not in allowed_capacities:
+                raise ValueError(
+                    f"Disk capacity must be one of: {allowed_capacities}. Got: {v}"
+                )
+        return v
+
+
+class DiskStatus(BaseModel):
+    status: Literal["Available", "Bound", "Released", "Failed", "Pending"]
+    capacityUsed: float | None = Field(
+        default=None, description="Used disk capacity in GB"
+    )
+    capacityAvailable: float | None = Field(
+        default=None, description="Available disk capacity in GB"
+    )
+    errorCode: str | None = Field(default=None, description="Error code if any")
+    diskBound: DiskBound | None = Field(
+        default=None, description="Disk bound information"
+    )
+
+
+class Disk(BaseModel):
+    """Disk model."""
+
+    apiVersion: str | None
+    kind: str | None
+    metadata: BaseMetadata = Field(description="Metadata for the Disk resource")
+    spec: DiskSpec = Field(description="Specification for the Disk resource")
+    status: DiskStatus | None = Field(default=None)
+
+
+class DiskList(BaseList[Disk]):
+    """List of disks using BaseList."""
+
+    pass

rapyuta_io_sdk_v2/pydantic_models/managedservice.py

@@ -0,0 +1,39 @@
+"""Pydantic models for ManagedService resource."""
+
+from typing import Any, Literal
+from pydantic import BaseModel, Field
+
+from rapyuta_io_sdk_v2.pydantic_models.utils import BaseMetadata, BaseList
+
+
+class ManagedServiceMetadata(BaseMetadata):
+    """Metadata for ManagedService resource."""
+
+    # Inherits all common fields from BaseMetadata
+    pass
+
+
+class ManagedServiceSpec(BaseModel):
+    """Specification for ManagedService resource."""
+
+    provider: Literal["elasticsearch", "headscalevpn"] = Field(
+        description="The provider for the managed service"
+    )
+    config: dict[str, Any] = Field(
+        description="Configuration object for the managed service"
+    )
+
+
+class ManagedService(BaseModel):
+    """Managed service model."""
+
+    apiVersion: str | None = None
+    kind: str | None = None
+    metadata: BaseMetadata
+    spec: ManagedServiceSpec
+
+
+class ManagedServiceList(BaseList[ManagedService]):
+    """List of managed services using BaseList."""
+
+    pass