1from pydantic import BaseModel, ConfigDict, Field, field_validator
2from typing import Optional, List, Dict, Literal, Any
3from enum import Enum
4
5
[docs]
6class OwnershipType(str, Enum):
7 PUBLIC = "Public"
8 OWNER_ONLY = "OwnerOnly"
9
10
[docs]
11class DesiredStatus(str, Enum):
12 RUNNING = "Running"
13 STOPPED = "Stopped"
14
15
[docs]
16class VolumeSpec(BaseModel):
17 """VolumeSpec defines a volume to mount from an existing PVC"""
18 name: str = Field(
19 description="Name is a unique identifier for this volume within the pod (maps to pod.spec.volumes[].name)",
20 min_length=1
21 )
22 mount_path: str = Field(
23 alias="mountPath",
24 description="MountPath is the path where the volume should be mounted (Unix-style path, e.g. /data)",
25 min_length=1
26 )
27 persistent_volume_claim_name: str = Field(
28 alias="persistentVolumeClaimName",
29 description="PersistentVolumeClaimName is the name of the existing PVC to mount",
30 min_length=1
31 )
32
33
[docs]
34class ContainerConfig(BaseModel):
35 """ContainerConfig defines container command and args configuration"""
36 command: Optional[List[str]] = Field(
37 default=None,
38 description="Command specifies the container command"
39 )
40 args: Optional[List[str]] = Field(
41 default=None,
42 description="Args specifies the container arguments"
43 )
44
45
[docs]
46class TemplateRef(BaseModel):
47 """TemplateRef defines a reference to a WorkspaceTemplate"""
48 name: str = Field(
49 description="Name of the WorkspaceTemplate"
50 )
51 namespace: Optional[str] = Field(
52 default=None,
53 description="Namespace where the WorkspaceTemplate is located"
54 )
55
56
[docs]
57class IdleDetectionSpec(BaseModel):
58 """IdleDetectionSpec defines idle detection methods"""
59 http_get: Optional[Dict[str, Any]] = Field(
60 default=None,
61 alias="httpGet",
62 description="HTTPGet specifies the HTTP request to perform for idle detection"
63 )
64
65
[docs]
66class IdleShutdownSpec(BaseModel):
67 """IdleShutdownSpec defines idle shutdown configuration"""
68 enabled: bool = Field(
69 description="Enabled indicates if idle shutdown is enabled"
70 )
71 idle_timeout_in_minutes: int = Field(
72 alias="idleTimeoutInMinutes",
73 description="IdleTimeoutInMinutes specifies idle timeout in minutes",
74 ge=1
75 )
76 detection: IdleDetectionSpec = Field(
77 description="Detection specifies how to detect idle state"
78 )
79
80
[docs]
81class StorageSpec(BaseModel):
82 """StorageSpec defines the storage configuration for Workspace"""
83 storage_class_name: Optional[str] = Field(
84 default=None,
85 alias="storageClassName",
86 description="StorageClassName specifies the storage class to use for persistent storage"
87 )
88 size: Optional[str] = Field(
89 default="10Gi",
90 description="Size specifies the size of the persistent volume. Supports standard Kubernetes resource quantities (e.g., '10Gi', '500Mi', '1Ti'). Integer values without units are interpreted as bytes"
91 )
92 mount_path: Optional[str] = Field(
93 default="/home",
94 alias="mountPath",
95 description="MountPath specifies where to mount the persistent volume in the container. Default is /home/jovyan (jovyan is the standard user in Jupyter images)"
96 )
97
98
[docs]
99class ResourceRequirements(BaseModel):
100 """ResourceRequirements describes the compute resource requirements"""
101 requests: Optional[Dict[str, Optional[str]]] = Field(
102 default=None,
103 description="Requests describes the minimum amount of compute resources required. If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, otherwise to an implementation-defined value. Requests cannot exceed Limits."
104 )
105 limits: Optional[Dict[str, Optional[str]]] = Field(
106 default=None,
107 description="Limits describes the maximum amount of compute resources allowed."
108 )
109
110
[docs]
111class SpaceConfig(BaseModel):
112 """SpaceConfig defines the desired state of a Space"""
113 model_config = ConfigDict(extra="forbid")
114
115 name: str = Field(
116 description="Space name",
117 min_length=1,
118 max_length=63,
119 pattern=r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?$'
120 )
121 display_name: str = Field(
122 alias="display_name",
123 description="Display Name of the space",
124 min_length=1
125 )
126 namespace: str = Field(
127 default="default",
128 description="Kubernetes namespace",
129 min_length=1
130 )
131 image: Optional[str] = Field(
132 default=None,
133 description="Image specifies the container image to use"
134 )
135 desired_status: Optional[DesiredStatus] = Field(
136 default=None,
137 alias="desired_status",
138 description="DesiredStatus specifies the desired operational status"
139 )
140 ownership_type: Optional[OwnershipType] = Field(
141 default=None,
142 alias="ownership_type",
143 description="OwnershipType specifies who can modify the space. 'Public' means anyone with RBAC permissions can update/delete the space. 'OwnerOnly' means only the creator can update/delete the space."
144 )
145 resources: Optional[ResourceRequirements] = Field(
146 default=None,
147 description="Resources specifies the resource requirements"
148 )
149 storage: Optional[StorageSpec] = Field(
150 default=None,
151 description="Storage specifies the storage configuration"
152 )
153 volumes: Optional[List[VolumeSpec]] = Field(
154 default=None,
155 description="Volumes specifies additional volumes to mount from existing PersistentVolumeClaims"
156 )
157 container_config: Optional[ContainerConfig] = Field(
158 default=None,
159 alias="container_config",
160 description="ContainerConfig specifies container command and args configuration"
161 )
162 node_selector: Optional[Dict[str, str]] = Field(
163 default=None,
164 alias="node_selector",
165 description="NodeSelector specifies node selection constraints for the space pod (JSON string)"
166 )
167 affinity: Optional[Dict[str, Any]] = Field(
168 default=None,
169 description="Affinity specifies node affinity and anti-affinity rules for the space pod (JSON string)"
170 )
171 tolerations: Optional[List[Dict[str, Any]]] = Field(
172 default=None,
173 description="Tolerations specifies tolerations for the space pod to schedule on nodes with matching taints (JSON string)"
174 )
175 lifecycle: Optional[Dict[str, Any]] = Field(
176 default=None,
177 description="Lifecycle specifies actions that the management system should take in response to container lifecycle events (JSON string)"
178 )
179 template_ref: Optional[TemplateRef] = Field(
180 default=None,
181 alias="template_ref",
182 description="TemplateRef references a WorkspaceTemplate to use as base configuration. When set, template provides defaults and workspace spec fields act as overrides"
183 )
184 idle_shutdown: Optional[IdleShutdownSpec] = Field(
185 default=None,
186 alias="idle_shutdown",
187 description="IdleShutdown specifies idle shutdown configuration"
188 )
189 app_type: Optional[str] = Field(
190 default=None,
191 alias="app_type",
192 description="AppType specifies the application type for this workspace"
193 )
194 service_account_name: Optional[str] = Field(
195 default=None,
196 alias="service_account_name",
197 description="ServiceAccountName specifies the name of the ServiceAccount to use for the workspace pod"
198 )
199
[docs]
200 @field_validator('volumes')
201 def validate_no_duplicate_volumes(cls, v):
202 """Validate no duplicate volume names or mount paths."""
203 if not v:
204 return v
205
206 # Check for duplicate volume names
207 names = [vol.name for vol in v]
208 if len(names) != len(set(names)):
209 raise ValueError("Duplicate volume names found")
210
211 # Check for duplicate mount paths
212 mount_paths = [vol.mount_path for vol in v]
213 if len(mount_paths) != len(set(mount_paths)):
214 raise ValueError("Duplicate mount paths found")
215
216 return v
217
[docs]
218 def to_domain(self) -> Dict:
219 """
220 Convert flat config to domain model for space creation
221 """
222 # Create the space spec
223 spec = {
224 "displayName": self.display_name
225 }
226
227 # Add optional spec fields
228 if self.image is not None:
229 spec["image"] = self.image
230 if self.desired_status is not None:
231 spec["desiredStatus"] = self.desired_status.value
232 if self.ownership_type is not None:
233 spec["ownershipType"] = self.ownership_type.value
234 if self.resources is not None:
235 spec["resources"] = self.resources.model_dump(exclude_none=True)
236 if self.storage is not None:
237 spec["storage"] = self.storage.model_dump(exclude_none=True, by_alias=True)
238 if self.volumes is not None:
239 spec["volumes"] = [vol.model_dump(exclude_none=True, by_alias=True) for vol in self.volumes]
240 if self.container_config is not None:
241 spec["containerConfig"] = self.container_config.model_dump(exclude_none=True)
242 if self.node_selector is not None:
243 spec["nodeSelector"] = self.node_selector
244 if self.affinity is not None:
245 spec["affinity"] = self.affinity
246 if self.tolerations is not None:
247 spec["tolerations"] = self.tolerations
248 if self.lifecycle is not None:
249 spec["lifecycle"] = self.lifecycle
250 if self.template_ref is not None:
251 spec["templateRef"] = self.template_ref.model_dump(exclude_none=True, by_alias=True)
252 if self.idle_shutdown is not None:
253 spec["idleShutdown"] = self.idle_shutdown.model_dump(exclude_none=True, by_alias=True)
254 if self.app_type is not None:
255 spec["appType"] = self.app_type
256 if self.service_account_name is not None:
257 spec["serviceAccountName"] = self.service_account_name
258
259 # Create metadata
260 metadata = {"name": self.name}
261 if self.namespace is not None:
262 metadata["namespace"] = self.namespace
263
264 # Create the complete space configuration
265 space_config = {
266 "apiVersion": "workspace.jupyter.org/v1alpha1",
267 "kind": "Workspace",
268 "metadata": metadata,
269 "spec": spec
270 }
271
272 return {
273 "name": self.name,
274 "namespace": self.namespace,
275 "space_spec": space_config
276 }
