VolumeDto

class VolumeDto(BaseModel)

A volume as returned by the API. last_manifest_id/conflicts are read best-effort from the CAS store on detail routes only (never on list).

Attributes:

• conflicts Optional[Any] = Field(default=None, description='Concurrent-write conflicts recorded on the latest manifest (audit trail).')
• created_at datetime
• error_reason Optional[StrictStr] = None
• id StrictStr
• last_manifest_id Optional[StrictStr] = Field(default=None, description='Blockmount CAS pointer, read-through on detail routes (best-effort).')
• last_used_at Optional[datetime] = None
• name StrictStr
• options Optional[Dict[str, Any]] = Field(default=None, description='Per-type options, namespaced by type (PLAN-VOLUME-TYPES.md §1): {} for a disk volume, {"bucket": {...}} for a bucket one. Credentials appear as the project-secret NAMES that were supplied, never as values — values are write-only across the whole API.')
• size_gb Optional[StrictInt] = Field(default=None, description='Scratch capacity in GiB. Disk volumes only — a bucket volume has no capacity of its own and reports null.')
• state StrictStr
• type StrictStr
• updated_at datetime
• used_bytes Optional[StrictInt] = Field(default=None, description='Real (logical) usage in bytes: sum of committed file sizes, hardlink groups counted once. Cached from runner events.volume.* reports; absent until the first report lands (never written / pre-feature volume).')
• used_bytes_at Optional[datetime] = Field(default=None, description='When the cached used_bytes was observed by the runner.')
• additional_properties Dict[str, Any] = {}

model_config

model_config = ConfigDict(populate_by_name=True, validate_assignment=True, protected_namespaces=())

VolumeDto.to_str

def to_str() -> str

Returns the string representation of the model using alias

VolumeDto.to_json

def to_json() -> str

Returns the JSON representation of the model using alias

VolumeDto.from_json

@classmethod
def from_json(json_str: str) -> Optional[Self]

Create an instance of VolumeDto from a JSON string

VolumeDto.to_dict

def to_dict() -> Dict[str, Any]

Return the dictionary representation of the model using alias.

This has the following differences from calling pydantic's self.model_dump(by_alias=True):

• None is only added to the output dict for nullable fields that were set at model initialization. Other fields with value None are ignored.
• Fields in self.additional_properties are added to the output dict.

VolumeDto.from_dict

@classmethod
def from_dict(obj: Optional[Dict[str, Any]]) -> Optional[Self]

Create an instance of VolumeDto from a dict

PullVolumeResult

@dataclass(frozen=True)
class PullVolumeResult()

Attributes:

• request_id str
• status PullVolumeStatus
• manifest_id Optional[str]
• raw_status Optional[str] = None

SandboxVolumesService

class SandboxVolumesService()

Pull attached volumes, separate from the root platform volume service.

SandboxVolumesService.__init__

def __init__(transport: AsyncTransport) -> None

SandboxVolumesService.pull

async def pull(volume_id: str,
               *,
               request_id: str,
               timeout: Optional[float] = None) -> PullVolumeResult

Submit/check once; timeout is a server wait in seconds (0–30, default 30).

Zero does not wait: it submits a new request_id, or returns a known one's last outcome without retrying it; a timeout above zero retries pending, unavailable or rate-limited work. Expiry does not cancel work. Neither retries nor identities are automatic.

VolumeService

class VolumeService()

Persistent volume namespace.

VolumeService.__init__

def __init__(transport: AsyncTransport) -> None

VolumeService.list

async def list(*, include_deleted: bool = False) -> ListResult[VolumeDto]

List volumes.

VolumeService.iter

async def iter(*, include_deleted: bool = False) -> AsyncIterator[VolumeDto]

VolumeService.get

async def get(name_or_id: str,
              *,
              create: bool = False,
              id_only: bool = False) -> VolumeDto

Fetch by id or name; id_only disables name fallback.

VolumeService.create

async def create(*, name: str, size_gb: Optional[int] = None) -> VolumeDto

Create a volume.

VolumeService.delete

async def delete(volume_id: str, *, id_only: bool = False) -> None

Request deletion and return once the control plane accepts it.

VolumeMount

@dataclass(frozen=True)
class VolumeMount()

Attach a persistent volume to a sandbox.

Attributes:

• volume_id str
• mount_path str
• subpath Optional[str] = None
• read_only Optional[bool] = None

SyncSandboxVolumesService

class SyncSandboxVolumesService()

Pull attached volumes, separate from the root platform volume service.

SyncSandboxVolumesService.__init__

def __init__(impl: SandboxVolumesService, bridge: SyncBridge) -> None

SyncSandboxVolumesService.pull

def pull(volume_id: str,
         *,
         request_id: str,
         timeout: Optional[float] = None) -> PullVolumeResult

Submit/check once; timeout is a server wait in seconds (0–30, default 30).

SyncVolumeService

class SyncVolumeService()

Persistent volume namespace.

SyncVolumeService.__init__

def __init__(impl: VolumeService, bridge: SyncBridge) -> None

SyncVolumeService.create

def create(*, name: str, size_gb: Optional[int] = None) -> VolumeDto

Create a volume.

SyncVolumeService.delete

def delete(volume_id: str, *, id_only: bool = False) -> None

Request deletion and return once the control plane accepts it.

SyncVolumeService.get

def get(name_or_id: str, *, create: bool = False, id_only: bool = False) -> VolumeDto

Fetch by id or name; id_only disables name fallback.

SyncVolumeService.iter

def iter(*, include_deleted: bool = False) -> Iterator[VolumeDto]

SyncVolumeService.list

def list(*, include_deleted: bool = False) -> ListResult[VolumeDto]

List volumes.

PullVolumeStatus

PullVolumeStatus = Literal['pulled', 'unchanged', 'pending', 'unknown']