> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreweave.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Sandbox
> Client CWSandbox doté d’une API hybride synchrone/asynchrone.
Source : [src/cwsandbox/\_sandbox.py:1688](https://github.com/coreweave/cwsandbox-client/blob/v1.14.2/src/cwsandbox/_sandbox.py#L1688)
```python theme={"system"}
class Sandbox(*, command: str | None = None, args: list[str] | None = None, defaults: SandboxDefaults | None = None, auth: AuthConfig | None = None, container_image: str | None = None, tags: list[str] | None = None, base_url: str | None = None, request_timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, max_lifetime_seconds: float | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, resources: ResourceOptions | dict[str, Any] | None = None, mounted_files: list[dict[str, Any]] | None = None, s3_mount: dict[str, Any] | None = None, ports: list[dict[str, Any]] | None = None, network: NetworkOptions | dict[str, Any] | None = None, file_system_snapshot: FileSystemSnapshotOptions | dict[str, Any] | None = None, max_timeout_seconds: int | None = None, placement_mode: PlacementMode | str | None = None, placement_spillover: PlacementSpillover | str | None = None, services: list[Service] | tuple[Service, ...] | None = None, volumes: list[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any]] | tuple[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any], ...] | None = None, template_id: str | None = None, image_pull_credentials: ImagePullCredentials | dict[str, Any] | None = None, runtime_class: str | None = None, security_context: SecurityContext | dict[str, Any] | None = None, working_dir: str | None = None, object_storage_access: ObjectStorageAccess | dict[str, Any] | None = None, environment_variables: dict[str, str] | None = None, annotations: dict[str, str] | None = None, secrets: Sequence[Secret | dict[str, Any]] | None = None, data_plane_mode: DataPlaneMode | str | None = None, containers: Sequence[Container | Mapping[str, Any]] | None = None)
```
Client CWSandbox doté d’une API hybride synchrone/asynchrone.
Toutes les méthodes rendent la main immédiatement et peuvent être utilisées dans des contextes synchrones comme asynchrones.
Les opérations sont exécutées dans une boucle d’événements en arrière-plan gérée par \_LoopManager.
Propriétés
sandbox\_id
```python theme={"system"}
@property
def sandbox_id(self) -> str | None
```
L’identifiant unique de la sandbox, ou None si elle n’a pas encore démarré.
returncode
```python theme={"system"}
@property
def returncode(self) -> int | None
```
Code de sortie si la sandbox est terminée, `None` si elle est toujours en cours d’exécution.
runner\_id
```python theme={"system"}
@property
def runner_id(self) -> str | None
```
Exécuteur sur lequel la sandbox s’exécute, ou None si elle n’a pas été démarrée.
status
```python theme={"system"}
@property
def status(self) -> SandboxStatus | None
```
Dernier statut connu de la sandbox.
status\_updated\_at
```python theme={"system"}
@property
def status_updated_at(self) -> datetime | None
```
Horodatage de la dernière confirmation du statut.
started\_at
```python theme={"system"}
@property
def started_at(self) -> datetime | None
```
Horodatage du démarrage de la sandbox.
runner\_group\_id
```python theme={"system"}
@property
def runner_group_id(self) -> str | None
```
ID du groupe d’exécuteurs dans lequel la sandbox s’exécute.
service\_urls
```python theme={"system"}
@property
def service_urls(self) -> tuple[tuple[int, str, str], ...]
```
URL propres à chaque service, attribuées par le backend.
service\_endpoints
```python theme={"system"}
@property
def service_endpoints(self) -> tuple[HttpsEndpointStatus, ...]
```
Points de terminaison HTTPS du produit renvoyés par les opérations create, Get ou list.
service\_addresses
```python theme={"system"}
@property
def service_addresses(self) -> tuple[TlsPassthroughEndpointStatus, ...]
```
Points de terminaison TLS passthrough renvoyés par les opérations create, Get ou list.
containers
```python theme={"system"}
@property
def containers(self) -> tuple[Container, ...]
```
Spécification du conteneur définie à la création, renvoyée telle quelle depuis la ressource sandbox.
container\_statuses
```python theme={"system"}
@property
def container_statuses(self) -> tuple[ContainerStatus, ...]
```
État observé de chaque conteneur. Le `status` de la sandbox reste déterminé par le conteneur principal.
dns\_egress\_names
```python theme={"system"}
@property
def dns_egress_names(self) -> tuple[str, ...]
```
Noms d’hôte autorisés à la création, renvoyés depuis `status.effective_egress`.
effective\_runtime\_class
```python theme={"system"}
@property
def effective_runtime_class(self) -> str | None
```
Classe de runtime appliquée par le backend, reprise du statut.
attached\_volume\_ids
```python theme={"system"}
@property
def attached_volume_ids(self) -> tuple[str, ...]
```
ID des Volumes enregistrés associés à cette sandbox, repris du statut.
effective\_egress
```python theme={"system"}
@property
def effective_egress(self) -> tuple[EgressRule, ...]
```
Règles de trafic sortant effectives, reprises de `status.effective_egress`.
effective\_ingress
```python theme={"system"}
@property
def effective_ingress(self) -> tuple[IngressRule, ...]
```
Règles d’ingress effectives, reprises de `status.effective_ingress`.
exposed\_ports
```python theme={"system"}
@property
def exposed_ports(self) -> tuple[tuple[int, str], ...] | None
```
Paires `(container_port, name)` exposées, dérivées des services typés.
resource\_limits
```python theme={"system"}
@property
def resource_limits(self) -> dict[str, str] | None
```
Limites de ressources renvoyées par la réponse de démarrage, ou None pour les bacs à sable découverts.
resource\_requests
```python theme={"system"}
@property
def resource_requests(self) -> dict[str, str] | None
```
Demandes de ressources issues de la réponse de démarrage, ou None pour les bacs à sable découverts.
resource\_gpu
```python theme={"system"}
@property
def resource_gpu(self) -> dict[str, Any] | None
```
Configuration GPU confirmée par la réponse de démarrage, ou None pour les bacs à sable découverts.
file\_system\_snapshot\_id
```python theme={"system"}
@property
def file_system_snapshot_id(self) -> str | None
```
ID de l’instantané créé par `stop(snapshot_on_stop=True)`.
exec\_stats
```python theme={"system"}
@property
def exec_stats(self) -> dict[str, int]
```
Statistiques d’exécution de cette sandbox.
Méthodes
run
```python theme={"system"}
run(*args: str = (), container_image: str | None = None, defaults: SandboxDefaults | None = None, auth: AuthConfig | None = None, request_timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, max_lifetime_seconds: float | None = None, tags: list[str] | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, resources: ResourceOptions | dict[str, Any] | None = None, mounted_files: list[dict[str, Any]] | None = None, s3_mount: dict[str, Any] | None = None, ports: list[dict[str, Any]] | None = None, network: NetworkOptions | dict[str, Any] | None = None, file_system_snapshot: FileSystemSnapshotOptions | dict[str, Any] | None = None, max_timeout_seconds: int | None = None, placement_mode: PlacementMode | str | None = None, placement_spillover: PlacementSpillover | str | None = None, services: list[Service] | tuple[Service, ...] | None = None, volumes: list[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any]] | tuple[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any], ...] | None = None, template_id: str | None = None, image_pull_credentials: ImagePullCredentials | dict[str, Any] | None = None, runtime_class: str | None = None, security_context: SecurityContext | dict[str, Any] | None = None, working_dir: str | None = None, object_storage_access: ObjectStorageAccess | dict[str, Any] | None = None, environment_variables: dict[str, str] | None = None, annotations: dict[str, str] | None = None, secrets: Sequence[Secret | dict[str, Any]] | None = None, data_plane_mode: DataPlaneMode | str | None = None, containers: Sequence[Container | Mapping[str, Any]] | None = None) -> Sandbox
```
Crée et démarre un sandbox, puis rend la main immédiatement dès que le backend accepte la requête.
N’attend PAS le statut RUNNING. Utilisez .wait() pour bloquer jusqu’à ce que le sandbox soit prêt.
Si des arguments positionnels sont fournis, le premier correspond à la commande et les suivants
à ses arguments. Si aucun argument n’est fourni, une commande de maintien en vie par défaut est utilisée : encapsulée dans un shell avec trap,
elle répond à SIGTERM lors de l’arrêt.
**Paramètres**
* `*args` (`str`) : commande et arguments facultatifs (par exemple, "echo", "hello", "world"). S’ils sont omis, la commande par défaut de SandboxDefaults est utilisée.
* `container_image` (`str | None`) : image de conteneur à utiliser
* `defaults` (`SandboxDefaults | None`) : SandboxDefaults facultatif à appliquer
* `auth` (`AuthConfig | None`) : mode ou fournisseur d’authentification. Remplace `defaults.auth`.
* `request_timeout_seconds` (`float | None`) : délai d’expiration des requêtes API (côté client)
* `poll_retry_budget_seconds` (`float | None`) : budget de temps réel alloué aux nouvelles tentatives en cas d’erreurs transitoires dans la boucle d’interrogation du statut de la sandbox (par défaut : 30 s). Définissez-le sur 0 pour désactiver les nouvelles tentatives.
* `poll_rpc_timeout_seconds` (`float | None`) : délai d’expiration par appel pour les RPC Get d’interrogation (par défaut : 15 s). Distinct de request\_timeout\_seconds.
* `max_lifetime_seconds` (`float | None`) : durée de vie maximale de la sandbox (côté serveur)
* `tags` (`list[str] | None`) : étiquettes facultatives pour la sandbox
* `profile_ids` (`list[str] | None`) : supprimé dans la version 1.x ; toute valeur transmise lève une `TypeError`.
* `profile_names` (`list[str] | None`) : supprimé dans la version 1.x ; toute valeur transmise lève une `TypeError`.
* `runner_ids` (`list[str] | None`) : épinglage facultatif sur un exécuteur CKS (incompatible avec le mode sans serveur et avec `placement_spillover='serverless_then_cks'`).
* `resources` (`ResourceOptions | dict[str, Any] | None`) : configuration des ressources. Accepte ResourceOptions pour définir séparément les demandes de ressources et les limites, ou un dict plat pour une QoS Guaranteed rétrocompatible.
* `mounted_files` (`list[dict[str, Any]] | None`) : fichiers à monter dans la sandbox au démarrage. Chaque dict doit contenir `mount_path` (str) et `file_content` (bytes). Remarque : les fichiers montés sont en lecture seule à l’exécution. Pour modifier un fichier, utilisez `sandbox.write_file()` une fois la sandbox en cours d’exécution.
* `s3_mount` (`dict[str, Any] | None`) : supprimé dans la version 1.x ; toute valeur transmise lève une `TypeError`.
* `ports` (`list[dict[str, Any]] | None`) : supprimé dans la version 1.x ; utilisez plutôt `services=[Service(...)]`.
* `network` (`NetworkOptions | dict[str, Any] | None`) : `NetworkOptions` (ou dict) avec des flags de refus et des autorisations facultatives de noms d’hôte définies à la création (`egress=[EgressRule(dns_name=...)]`). L’exposition des ports passe par `services=`.
* `placement_mode` (`PlacementMode | str | None`) : `PlacementMode` ou chaîne (`serverless` / `cks`).
* `placement_spillover` (`PlacementSpillover | str | None`) : `PlacementSpillover` ou chaîne. Par défaut : `strict`. Voir `Sandbox.__init__`.
* `services` (`list[Service] | tuple[Service, ...] | None`) : ports de service typés (liste/tuple de `Service`).
* `volumes` (`list[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any]] | tuple[ScratchVolumeOptions | RegisteredVolumeOptions | dict[str, Any], ...] | None`) : volumes temporaires ou enregistrés (`ScratchVolumeOptions`, `RegisteredVolumeOptions` ou un dict `volume_id`).
* `runtime_class` (`str | None`) : épinglage facultatif d’une classe de runtime (par exemple, `"gvisor"`).
* `security_context` (`SecurityContext | dict[str, Any] | None`) : privilèges dans l’invité pour le conteneur principal.
* `working_dir` (`str | None`) : répertoire de travail de la commande du conteneur principal.
* `object_storage_access` (`ObjectStorageAccess | dict[str, Any] | None`) : identifiants d’authentification temporaires pour le stockage d’objets.
* `file_system_snapshot` (`FileSystemSnapshotOptions | dict[str, Any] | None`) : options FSS simplifiées pour un montage unique (`FileSystemSnapshotOptions` ou dict). Privilégiez `volumes=` pour les configurations à plusieurs volumes.
* `max_timeout_seconds` (`int | None`) : supprimé dans la version 1.x ; utilisez `request_timeout_seconds`.
* `environment_variables` (`dict[str, str] | None`) : variables d’environnement à injecter dans la sandbox. Fusionnées avec les valeurs par défaut de la session, dont elles remplacent les clés correspondantes. À réserver à une configuration non sensible.
* `annotations` (`dict[str, str] | None`) : annotations de pod Kubernetes pour la sandbox. Fusionnées avec les valeurs par défaut de la session, dont elles remplacent les clés correspondantes. À réserver à des métadonnées non sensibles.
* `secrets` (`Sequence[Secret | dict[str, Any]] | None`) : secrets à injecter en tant que variables d’environnement lors de la création. Fusionnés avec les valeurs par défaut (d’abord les valeurs par défaut, puis cette liste).
* `data_plane_mode` (`DataPlaneMode | str | None`) : politique de transport pour l’exécution, les journaux et les opérations sur les fichiers. `auto` privilégie le mTLS direct, avec repli sur la passerelle.
* `containers` (`Sequence[Container | Mapping[str, Any]] | None`) : spécification multi-conteneurs. Incompatible avec les arguments positionnels command/args ainsi qu’avec `container_image`, `resources`, `mounted_files`, `secrets`, `image_pull_credentials`, `environment_variables`, `security_context` et `working_dir`. Cette liste remplace ces champs propres à un conteneur unique, y compris les champs de même nom dans `SandboxDefaults`. Définissez les secrets, les variables d’environnement et working\_dir dans chaque `Container`.
Retourne :
Une instance de Sandbox (la requête de démarrage a été envoyée, mais le sandbox est peut-être encore en cours de démarrage)
**Exemples**
```python theme={"system"}
# Avec les valeurs par défaut (maintien actif via un trap du shell)
sb = Sandbox.run()
# Mode « fire and forget » (lancer sans attendre)
sb = Sandbox.run("echo", "hello")
# sb.sandbox_id est défini, mais la sandbox est peut-être encore en cours de démarrage
# Attendre que la sandbox soit prête si nécessaire
sb = Sandbox.run("sleep", "infinity").wait()
result = sb.exec(["echo", "hello"]).result()
# Ou utiliser un gestionnaire de contexte pour un nettoyage automatique
with Sandbox.run("sleep", "infinity") as sb:
result = sb.exec(["echo", "hello"]).result()
```
run\_from\_template
```python theme={"system"}
run_from_template(template_id: str, *args: str = (), command: str | None = None, defaults: SandboxDefaults | None = None, **kwargs: Any = {}) -> Sandbox
```
Crée une sandbox à partir d’un modèle d’organisation (CreateSandboxFromTemplate).
**Paramètres**
* `template_id` (`str`) : UUID du modèle, limité à l’organisation.
* `*args` (`str`) : remplacement facultatif des arguments de la commande. Pris en compte en l’absence de `command`. Les surcouches partielles d’un conteneur unique nécessitent toujours `container_image`.
* `command` (`str | None`) : remplacement facultatif de la commande. Nécessite `container_image`, sauf si `containers=` remplace la liste entière.
* `defaults` (`SandboxDefaults | None`) : SandboxDefaults facultatif.
* `**kwargs` (`Any`) : mêmes kwargs de création avancés que `run()`. Le passage de `container_image` remplace l’intégralité du conteneur du modèle (commande, arguments, variables d’environnement, fichiers, ressources, nom `main`) ; le conteneur existant n’est ni récupéré ni fusionné. `containers=` remplace la liste complète et ne nécessite pas `container_image`. Les autres surcouches de champs de conteneur (`command`, `args`, `environment_variables`, `secrets`, `resources`, `mounted_files`, `volumes`, `file_system_snapshot`, `image_pull_credentials`, `security_context`, `working_dir`) nécessitent `container_image`, sauf si `containers=` remplace la liste : l’API remplace la liste entière des conteneurs et rejette tout patch partiel. Les tags de session et par défaut sont fusionnés puis envoyés sous forme de remplacement appliqué dès qu’ils sont présents, afin que `list()`/`adopt` puissent trouver la sandbox ; les variables d’environnement et les annotations ne sont pas fusionnées (elles relèvent du modèle).
**Renvoie**
* `Sandbox` : handle de la sandbox (démarrage différé).
run\_from\_file
```python theme={"system"}
run_from_file(contents: str | Path | bytes, *, primary_service: str, file_type: SandboxFileType | str = SandboxFileType.COMPOSE, image_overrides: Mapping[str, str] | None = None, default_resources: ResourceOptions | dict[str, Any] | None = None, defaults: SandboxDefaults | None = None, auth: AuthConfig | None = None, request_timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, max_lifetime_seconds: float | None = None, tags: list[str] | None = None, runner_ids: list[str] | None = None, network: NetworkOptions | dict[str, Any] | None = None, placement_mode: PlacementMode | str | None = None, placement_spillover: PlacementSpillover | str | None = None, object_storage_access: ObjectStorageAccess | dict[str, Any] | None = None, annotations: dict[str, str] | None = None, request_id: str | None = None, data_plane_mode: DataPlaneMode | str | None = None, **kwargs: Any = {}) -> Sandbox
```
Crée une sandbox à partir d’un fichier téléversé (CreateSandboxFromFile).
v1 : Compose uniquement, images à récupérer uniquement (pull). Lit `contents` sous forme d’octets bruts et
ne normalise pas le YAML : un même `request_id` associé à des octets différents
(y compris des modifications portant uniquement sur les caractères d’espacement) produit `CWSANDBOX_REQUEST_ID_CONFLICT`.
Rend la main dès que le backend accepte la requête ; utilisez `wait()` pour attendre
RUNNING. N’attendez pas `PREPARING`.
GetSandbox renvoie la spécification traduite, et non le fichier source. Une directive `build:`
Compose résiduelle produit `CWSANDBOX_NOT_IMPLEMENTED`. Pour éviter la compilation,
définissez `image:` dans le YAML ou utilisez `image_overrides`.
**Paramètres**
* `contents` (`str | Path | bytes`) : chemin du fichier Compose (`str` / `Path`) ou octets YAML bruts. Une `str` est toujours interprétée comme un chemin ; pour transmettre du texte Compose, passez-le sous forme de `bytes` (`.encode("utf-8")`). Taille plafonnée à 256 Kio. Les octets sont envoyés tels quels.
* `primary_service` (`str`) : service Compose qui joue le rôle de service principal de la sandbox.
* `file_type` (`SandboxFileType | str`) : type de document. Valeur par défaut : `compose`.
* `image_overrides` (`Mapping[str, str] | None`) : réf. d’images récupérables, par service. Les clés doivent correspondre à des services définis dans `contents`.
* `default_resources` (`ResourceOptions | dict[str, Any] | None`) : CPU/mémoire appliqués à chaque service qui ne définit pas `deploy.resources`. S’applique par conteneur, et non comme un budget de projet. Le GPU est rejeté localement (la passerelle le rejette également).
* `defaults` (`SandboxDefaults | None`) : `SandboxDefaults` facultatif. Les tags, le réseau, `placement_mode`, `runner_ids`, les annotations, l’accès au stockage d’objets et la durée de vie maximale sont hérités. Les valeurs par défaut de conteneur, de volume et de service ne le sont pas. Un `placement_spillover` non strict dans les valeurs par défaut est ignoré (la création à partir d’un fichier est toujours stricte). Un argument nommé explicite non strict lève une exception.
* `request_id` (`str | None`) : jeton d’idempotence facultatif. Généré automatiquement s’il est omis.
* `**kwargs` (`Any`) : rejetés. Ce RPC n’accepte ni les volumes, ni les services publiés, ni `runtime_class`, `image_pull_credentials`, `network_ids` ou `build_contexts`.
**Renvoie**
* `Sandbox` : handle de la sandbox (création acceptée ; la sandbox peut être encore en cours de démarrage).
session
```python theme={"system"}
session(defaults: SandboxDefaults | Mapping[str, Any] | None = None, *, auth: AuthConfig | None = None) -> Session
```
Crée une session pour gérer plusieurs bacs à sable.
Les sessions offrent :
* Une configuration partagée grâce aux valeurs par défaut
* Le nettoyage automatique des bacs à sable orphelins
* L’exécution de fonctions grâce au décorateur @session.function()
**Paramètres**
* `defaults` (`SandboxDefaults | Mapping[str, Any] | None`) : valeurs par défaut facultatives à appliquer aux bacs à sable créés par la session
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur. S’il est fourni, ce paramètre remplace `defaults.auth`.
**Renvoie**
* `Session` : une instance de Session
**Exemples**
```python theme={"system"}
session = Sandbox.session(defaults)
sb = session.create(command="sleep", args=["infinity"])
@session.function()
def compute(x, y):
return x + y
await session.close()
```
list
```python theme={"system"}
list(*, tags: list[str] | None = None, status: str | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, show_terminated: bool = False, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, data_plane_mode: DataPlaneMode | str = DataPlaneMode.AUTO, volume_ids: list[str] | tuple[str, ...] | None = None) -> OperationRef[list[Sandbox]]
```
Liste les bacs à sable existants, avec des filtres facultatifs.
Renvoie un OperationRef qui se résout en instances de Sandbox utilisables pour
des opérations telles que exec(), stop(), get\_status(), read\_file(), write\_file().
Par défaut, seuls les bacs à sable actifs (non terminaux) sont renvoyés.
Définissez `show_terminated=True` pour élargir la recherche aux bacs à sable
terminaux (terminés, en échec, arrêtés).
Un filtre sur un statut terminal (par exemple `status="completed"`) élargit également
la recherche automatiquement.
**Paramètres**
* `tags` (`list[str] | None`) : filtre par tags (les bacs à sable doivent posséder TOUS les tags spécifiés)
* `status` (`str | None`) : filtre par statut ("running", "completed", "failed", etc.)
* `profile_ids` (`list[str] | None`) : supprimé dans la version 1.x ; toute valeur transmise lève une `TypeError`.
* `profile_names` (`list[str] | None`) : supprimé dans la version 1.x ; toute valeur transmise lève une `TypeError`.
* `runner_ids` (`list[str] | None`) : filtre par ID d’exécuteur
* `volume_ids` (`list[str] | tuple[str, ...] | None`) : limite les résultats aux bacs à sable associés à ces ID de Volumes enregistrés
* `show_terminated` (`bool`) : si True, inclut les bacs à sable terminaux (terminés, en échec, arrêtés). Valeur par défaut : False.
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut)
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur à utiliser pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s)
* `poll_retry_budget_seconds` (`float | None`) : budget de temps réel alloué aux nouvelles tentatives en cas d’erreurs transitoires dans la boucle d’interrogation du statut du sandbox (par défaut : 30 s). Définissez 0 pour désactiver les nouvelles tentatives. S’applique aux instances de Sandbox renvoyées.
* `poll_rpc_timeout_seconds` (`float | None`) : délai d’expiration par appel pour les RPC Get d’interrogation (par défaut : 15 s). Distinct de `timeout_seconds`. S’applique aux instances de Sandbox renvoyées.
* `data_plane_mode` (`DataPlaneMode | str`) : politique de transport appliquée aux bacs à sable renvoyés.
**Renvoie**
* `OperationRef[list[Sandbox]]` : OperationRef\[list\[Sandbox]] : utilisez .result() pour attendre les résultats de manière bloquante,
* `OperationRef[list[Sandbox]]` : ou utilisez directement await dans les contextes asynchrones.
**Exemples**
```python theme={"system"}
# Utilisation synchrone - bacs à sable actifs uniquement (par défaut)
sandboxes = Sandbox.list(tags=["my-batch-job"]).result()
for sb in sandboxes:
print(f"{sb.sandbox_id}: {sb.status}")
sb.stop().result()
# Inclure les bacs à sable arrêtés
all_sandboxes = Sandbox.list(
tags=["my-batch-job"], show_terminated=True
).result()
# Utilisation asynchrone
sandboxes = await Sandbox.list(status="running")
for sb in sandboxes:
result = await sb.exec(["echo", "hello"])
```
from\_id
```python theme={"system"}
from_id(sandbox_id: str, *, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, data_plane_mode: DataPlaneMode | str = DataPlaneMode.AUTO) -> OperationRef[Sandbox]
```
Se rattacher à une sandbox existante à partir de son ID.
Crée une instance de Sandbox connectée à une sandbox existante,
ce qui permet d’effectuer des opérations comme exec(), stop(), get\_status(), etc.
**Paramètres**
* `sandbox_id` (`str`) : l’ID de la sandbox existante
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut)
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s)
* `poll_retry_budget_seconds` (`float | None`) : budget de temps réel alloué aux nouvelles tentatives en cas d’erreurs transitoires dans la boucle d’interrogation du statut de la sandbox (par défaut : 30 s). Définissez-le sur 0 pour désactiver les nouvelles tentatives. S’applique à l’instance de Sandbox renvoyée.
* `poll_rpc_timeout_seconds` (`float | None`) : délai d’expiration par appel pour les RPC Get d’interrogation (par défaut : 15 s). Distinct de `timeout_seconds`. S’applique à l’instance de Sandbox renvoyée.
* `data_plane_mode` (`DataPlaneMode | str`) : politique de transport appliquée à la sandbox renvoyée.
**Renvoie**
* `OperationRef[Sandbox]` : OperationRef\[Sandbox] : utilisez .result() pour obtenir l’instance de Sandbox de manière bloquante,
* `OperationRef[Sandbox]` : ou utilisez directement await dans les contextes asynchrones.
**Exceptions levées**
* `SandboxNotFoundError` : si la sandbox n'existe pas
**Exemples**
```python theme={"system"}
# Utilisation synchrone
sb = Sandbox.from_id("sandbox-abc123").result()
result = sb.exec(["python", "-c", "print('hello')"]).result()
sb.stop().result()
# Utilisation asynchrone
sb = await Sandbox.from_id("sandbox-abc123")
result = await sb.exec(["python", "-c", "print('hello')"])
```
delete
```python theme={"system"}
delete(sandbox_id: str, *, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None, missing_ok: bool = False) -> OperationRef[None]
```
Supprime une sandbox à partir de son ID sans créer d’instance Sandbox.
Cette méthode utilitaire est pratique pour les scénarios de nettoyage dans lesquels vous
n'avez pas besoin d’effectuer d’autres opérations sur la sandbox.
**Paramètres**
* `sandbox_id` (`str`) : ID de la sandbox à supprimer
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut)
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s)
* `missing_ok` (`bool`) : si True, ignore l’erreur SandboxNotFoundError lorsque la sandbox n'existe pas.
**Renvoie**
* `OperationRef[None]` : OperationRef\[None] : utilisez .result() pour attendre la fin de l’opération.
* `OperationRef[None]` : lève SandboxNotFoundError si la sandbox est introuvable (sauf si missing\_ok=True),
* `OperationRef[None]` : SandboxError si la suppression a échoué.
**Exceptions levées**
* `SandboxNotFoundError` : si la sandbox n'existe pas et que missing\_ok=False
* `SandboxError` : si la suppression a échoué pour une autre raison
**Exemples**
```python theme={"system"}
# Utilisation synchrone
Sandbox.delete("sandbox-abc123").result()
# Ignorer si la sandbox est déjà supprimée
Sandbox.delete("sandbox-abc123", missing_ok=True).result()
# Utilisation asynchrone
await Sandbox.delete("sandbox-abc123")
```
get\_snapshot
```python theme={"system"}
get_snapshot(file_system_snapshot_id: str, *, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None) -> OperationRef[FileSystemSnapshot]
```
Récupère un enregistrement d’instantané de système de fichiers (FSS) par son ID.
Les instantanés sont limités à l’organisation : tout instantané appartenant à votre organisation est
visible, quelle que soit la sandbox qui l’a créé.
**Paramètres**
* `file_system_snapshot_id` (`str`) : ID de l’instantané à récupérer.
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut).
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s).
**Renvoie**
* `OperationRef[FileSystemSnapshot]` : OperationRef\[FileSystemSnapshot] : utilisez .result() pour un appel bloquant, ou await.
* `OperationRef[FileSystemSnapshot]` : lève SnapshotNotFoundError si l’instantané n’existe pas.
**Exemples**
```python theme={"system"}
snap = Sandbox.get_snapshot("fss-abc123").result()
print(snap.status, snap.size_bytes)
```
list\_snapshots
```python theme={"system"}
list_snapshots(*, source_sandbox_id: str | None = None, status: FileSystemSnapshotStatus | str | None = None, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None) -> OperationRef[list[FileSystemSnapshot]]
```
Liste les instantanés de système de fichiers (FSS) de l’organisation.
Les instantanés sont limités à l’organisation et la liste est paginée automatiquement. Les
filtres `source_sandbox_id` et `status` sont appliqués côté client
(le RPC de liste du backend ne filtre pas) : tous les instantanés sont donc récupérés
avant le filtrage.
**Paramètres**
* `source_sandbox_id` (`str | None`) : si défini, seuls les instantanés capturés à partir de cette sandbox sont renvoyés.
* `status` (`FileSystemSnapshotStatus | str | None`) : si défini, seuls les instantanés ayant ce statut sont renvoyés (FileSystemSnapshotStatus ou sa valeur string).
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut).
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s).
**Renvoie**
* `OperationRef[list[FileSystemSnapshot]]` : OperationRef\[list\[FileSystemSnapshot]] : utilisez .result() pour un appel bloquant, ou await.
**Exemples**
```python theme={"system"}
# Tous les snapshots à l'état prêt issus d'une sandbox donnée
snaps = Sandbox.list_snapshots(
source_sandbox_id=sb.sandbox_id,
status=FileSystemSnapshotStatus.READY,
).result()
```
delete\_snapshot
```python theme={"system"}
delete_snapshot(file_system_snapshot_id: str, *, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None, missing_ok: bool = False) -> OperationRef[None]
```
Supprime un instantané de système de fichiers (FSS) à partir de son ID.
La suppression d’un instantané n’affecte pas les bacs à sable déjà restaurés à partir de celui-ci.
**Paramètres**
* `file_system_snapshot_id` (`str`) : ID de l’instantané à supprimer.
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut).
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s).
* `missing_ok` (`bool`) : si True, ignore l’exception SnapshotNotFoundError lorsque l’instantané n'existe pas (déjà supprimé).
**Renvoie**
* `OperationRef[None]` : OperationRef\[None] : utilisez .result() pour un appel bloquant, ou await.
* `OperationRef[None]` : lève SnapshotNotFoundError si l’instantané est introuvable (sauf si missing\_ok=True).
**Exemples**
```python theme={"system"}
Sandbox.delete_snapshot("fss-abc123").result()
Sandbox.delete_snapshot("fss-abc123", missing_ok=True).result()
```
get\_snapshot\_bucket\_config
```python theme={"system"}
get_snapshot_bucket_config(*, base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None) -> OperationRef[FileSystemSnapshotBucketConfig]
```
Récupère la configuration du bucket de stockage d'objets FSS de l'organisation.
**Paramètres**
* `base_url` (`str | None`) : Remplacer l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut).
* `auth` (`AuthConfig | None`) : Stratégie d’authentification, en-têtes résolus ou fournisseur à utiliser pour cette requête.
* `timeout_seconds` (`float | None`) : Délai d’expiration de la requête (par défaut : 300 s).
**Renvoie**
* `OperationRef[FileSystemSnapshotBucketConfig]` : OperationRef\[FileSystemSnapshotBucketConfig] : utilisez .result() ou await.
**Exemples**
```python theme={"system"}
cfg = Sandbox.get_snapshot_bucket_config().result()
print(cfg.mode, cfg.effective_bucket_name)
```
set\_snapshot\_bucket\_config
```python theme={"system"}
set_snapshot_bucket_config(*, bucket_name: str, region: str = '', base_url: str | None = None, auth: AuthConfig | None = None, timeout_seconds: float | None = None) -> OperationRef[FileSystemSnapshotBucketConfig]
```
Définit la configuration du bucket de stockage d'objets FSS de l'organisation.
Fournissez un `bucket_name` pour utiliser votre propre bucket ; passez une chaîne
vide pour revenir au bucket managé par CoreWeave. Cette opération est
réservée aux administrateurs.
**Paramètres**
* `bucket_name` (`str`) : bucket dans lequel archiver les instantanés. Une chaîne vide rétablit le bucket managé par CoreWeave.
* `region` (`str`) : région du bucket (requise par certains fournisseurs pour les buckets BYO).
* `base_url` (`str | None`) : remplace l’URL de l’API (par défaut : variable d’environnement CWSANDBOX\_BASE\_URL ou valeur par défaut).
* `auth` (`AuthConfig | None`) : stratégie d’authentification, en-têtes résolus ou fournisseur à utiliser pour cette requête.
* `timeout_seconds` (`float | None`) : délai d’expiration de la requête (par défaut : 300 s).
**Renvoie**
* `OperationRef[FileSystemSnapshotBucketConfig]` : OperationRef\[FileSystemSnapshotBucketConfig] : la configuration mise à jour.
**Exemples**
```python theme={"system"}
# Utiliser votre propre bucket
Sandbox.set_snapshot_bucket_config(
bucket_name="my-org-fss", region="us-east-1"
).result()
# Revenir au bucket managé par CoreWeave
Sandbox.set_snapshot_bucket_config(bucket_name="").result()
```
get\_status
```python theme={"system"}
get_status() -> SandboxStatus
```
Obtient le statut actuel de la sandbox.
Pour les sandboxes dans un état terminal (COMPLETED/FAILED/TERMINATED), renvoie le statut
mis en cache sans appel d’API. Pour les sandboxes actives, récupère le statut depuis le backend.
**Renvoie**
* `SandboxStatus` : valeur de l’énumération SandboxStatus
**Exceptions levées**
* `SandboxNotRunningError` : si la sandbox n’a pas été démarrée
**Exemples**
```python theme={"system"}
sb = Sandbox.run("sleep", "10")
status = sb.get_status()
print(f"Sandbox is {status}") # SandboxStatus.PENDING ou RUNNING
```
start
```python theme={"system"}
start() -> OperationRef[None]
```
Envoie StartSandbox au backend et renvoie immédiatement un OperationRef.
N’attend PAS le statut RUNNING. Utilisez wait() pour bloquer l’exécution jusqu’à ce que la sandbox soit prête.
Appelez .result() pour bloquer l’exécution jusqu’à ce que la requête de démarrage soit acceptée.
**Renvoie**
* `OperationRef[None]` : OperationRef\[None] : utilisez .result() pour bloquer l’exécution jusqu’à ce que le backend accepte la requête.
**Exemples**
```python theme={"system"}
sandbox = Sandbox(command="sleep", args=["infinity"])
sandbox.start().result()
print(f"Started sandbox: {sandbox.sandbox_id}")
sandbox.wait() # Bloque jusqu’à l’état RUNNING
```
wait
```python theme={"system"}
wait(timeout: float | None = None) -> Sandbox
```
Bloque jusqu’à ce que la sandbox atteigne l’état RUNNING ou un état terminal.
Rend la main lorsque la sandbox est à l’état RUNNING ou est déjà terminée (COMPLETED/UNSPECIFIED).
**Paramètres**
* `timeout` (`float | None`) : Nombre maximal de secondes d’attente. None signifie que le délai d’expiration par défaut est utilisé.
**Renvoie**
* `Sandbox` : L’instance elle-même, pour permettre le chaînage de méthodes. Consultez .status pour déterminer l’état final.
**Exceptions levées**
* `SandboxFailedError` : Si la sandbox ne parvient pas à démarrer
* `SandboxTerminatedError` : Si la sandbox a été arrêtée de l’extérieur
* `SandboxTimeoutError` : Si le délai d’expiration est dépassé
**Exemples**
```python theme={"system"}
sb = Sandbox.run("sleep", "infinity").wait()
result = sb.exec(["echo", "ready"]).result()
```
wait\_until\_complete
```python theme={"system"}
wait_until_complete(timeout: float | None = None, *, raise_on_termination: bool = True) -> OperationRef[Sandbox]
```
Attend que la sandbox atteigne un état terminal (COMPLETED/FAILED/TERMINATED).
Renvoie un OperationRef qui se résout lorsque la sandbox atteint un état terminal.
Une fois la résolution effectuée, returncode est disponible si le backend
en a enregistré un (consultez la propriété `returncode` pour connaître les cas où il
reste None).
**Paramètres**
* `timeout` (`float | None`) : durée d’attente maximale, en secondes. None signifie que le délai d’expiration par défaut s’applique.
* `raise_on_termination` (`bool`) : si True (valeur par défaut), lève SandboxTerminatedError lorsque ce client a appelé stop() ou que le backend signale l’ancien statut TERMINATED. Les arrêts externes (infrastructure, limites de durée de vie, autres clients) qui aboutissent à COMPLETED ne peuvent pas être détectés tant que le backend ne fournit pas la métadonnée termination\_reason. Définissez la valeur sur False pour ne jamais lever SandboxTerminatedError.
**Renvoie**
* `OperationRef[Sandbox]` : OperationRef\[Sandbox] : utilisez .result() pour bloquer, ou await dans les contextes asynchrones.
**Exceptions levées**
* `SandboxTimeoutError` : si le délai d’expiration est dépassé
* `SandboxTerminatedError` : si la sandbox a été arrêtée par ce client ou signalée comme TERMINATED par le backend (et que raise\_on\_termination=True)
* `SandboxFailedError` : si la sandbox a échoué
**Remarque :** `poll_retry_budget_seconds` est un sous-délai d’expiration strict, compris dans le paramètre `timeout` de l’utilisateur. Avec un budget de nouvelles tentatives de 30 s et un délai d’expiration utilisateur de 300 s, des erreurs d’épuisement du budget peuvent survenir au bout d’environ 30 s. Si vous souhaitez prolonger les nouvelles tentatives, configurez `poll_retry_budget_seconds` en conséquence.
**Exemples**
```python theme={"system"}
sb = Sandbox.run("python", "-c", "print('done')")
sb.wait_until_complete().result()
print(f"Exit code: {sb.returncode}")
```
stop
```python theme={"system"}
stop(*, snapshot_on_stop: bool = False, graceful_shutdown_seconds: float = DEFAULT_GRACEFUL_SHUTDOWN_SECONDS, missing_ok: bool = False, wait_for_ready: bool = True, request_id: str | None = None) -> OperationRef[None]
```
Arrête la sandbox et renvoie immédiatement un OperationRef.
La sandbox passe par l’état TERMINATING (drainage pendant le délai de grâce)
avant d’atteindre un état terminal (COMPLETED ou FAILED). L’OperationRef
renvoyé est résolu lorsque le backend confirme un état terminal, et pas
seulement lorsque l’appel RPC d’arrêt réussit.
Plusieurs appelants partagent la même tâche d’arrêt sous-jacente : le premier appelant
la crée, les suivants s’y rattachent. Une requête `snapshot_on_stop=True`
qui se rattacherait à (ou observerait) un arrêt ne capturant pas de
snapshot, parce que la sandbox est déjà en cours d’arrêt, déjà arrêtée, ou
qu’un simple `stop()` est déjà en cours, lève
`SnapshotOnStopConflictError` au lieu de se terminer silencieusement sans
archive. Les arrêts simples sont toujours regroupés.
La sandbox est désinscrite de sa session, que l’arrêt
ait réussi ou non, car elle n’est plus utilisable.
**Paramètres**
* `snapshot_on_stop` (`bool`) : si True, capture un instantané du système de fichiers (FSS) du point de montage configuré avant l’arrêt. L’ID du snapshot obtenu est accessible via la propriété `file_system_snapshot_id` une fois l’OperationRef renvoyé résolu. Nécessite que la sandbox ait été démarrée avec un point de montage `file_system_snapshot` et que FSS soit activé pour l’organisation. Lève `SnapshotOnStopConflictError` si un arrêt qui ne capturera pas de snapshot est déjà en cours.
* `graceful_shutdown_seconds` (`float`) : délai d’attente pour l’arrêt en douceur. Avec `snapshot_on_stop=True`, il s’agit du délai de grâce de suppression du pod après l’archivage, appliqué *après* la fin du snapshot ; l’échéance côté client couvre donc le budget d’archivage plus ce délai de grâce. En v1, `grace_period_seconds=0` signifie un arrêt immédiat (aucun délai de grâce de substitution côté backend). Le backend plafonne cette valeur à 300 s pour snapshot-on-stop.
* `missing_ok` (`bool`) : si True, supprime SandboxNotFoundError lorsque la sandbox n’existe pas. Avec `snapshot_on_stop=True`, l’option est tout de même envoyée sous la forme `allow_missing` ; une sandbox introuvable ou un rejet de cette combinaison par le serveur lève `SnapshotOnStopConflictError` au lieu de réussir sans archive.
* `wait_for_ready` (`bool`) : lorsque `snapshot_on_stop` vaut True, bloque jusqu’à ce que le snapshot atteigne READY (ou FAILED) avant la fin de l’arrêt. Ignoré lorsque `snapshot_on_stop` vaut False.
* `request_id` (`str | None`) : clé facultative fournie par le client pour dédupliquer la requête snapshot-on-stop lors des nouvelles tentatives. Ignoré lorsque `snapshot_on_stop` vaut False.
**Renvoie**
* `OperationRef[None]` : OperationRef\[None] : utilisez .result() pour bloquer jusqu’à l’état terminal.
* `OperationRef[None]` : lève SandboxError en cas d’échec, SandboxNotFoundError si introuvable
* `OperationRef[None]` : (sauf si missing\_ok=True).
**Exemples**
```python theme={"system"}
sb.stop().result() # Bloque jusqu’à un état terminal (COMPLETED/FAILED)
# Ignorer si déjà supprimé
sb.stop(missing_ok=True).result()
# Créer un snapshot du montage configuré lors de l’arrêt, puis lire l’ID
sb.stop(snapshot_on_stop=True).result()
file_system_snapshot_id = sb.file_system_snapshot_id
# Après stop(), wait_until_complete() se résout une fois l’état terminal atteint
sb.stop()
sb.wait_until_complete().result() # Interroge l’état tout au long de TERMINATING
```
snapshot
```python theme={"system"}
snapshot(*, wait_for_ready: bool = True, request_id: str | None = None) -> OperationRef[str]
```
Capture un instantané du système de fichiers (FSS) du point de montage configuré.
Crée un instantané du répertoire configuré via `file_system_snapshot` sur la
sandbox en cours d’exécution, sans l’arrêter. Démarre d’abord la sandbox si elle
n’a pas encore été démarrée. Restaurez l’instantané dans une nouvelle sandbox via
`Sandbox.run(file_system_snapshot=FileSystemSnapshotOptions(...,
file_system_snapshot_id=))`.
Nécessite que la sandbox ait été démarrée avec un point de montage `file_system_snapshot`
et que FSS soit activé pour l’organisation.
**Paramètres**
* `wait_for_ready` (`bool`) : bloque jusqu’à ce que l’instantané atteigne l’état READY (ou FAILED) avant de rendre la main. Si False, rend la main dès que l’instantané est créé (probablement encore à l’état CREATING).
* `request_id` (`str | None`) : clé facultative fournie par le client pour dédupliquer la requête en cas de nouvelles tentatives.
**Renvoie**
* `OperationRef[str]` : OperationRef\[str] : utilisez .result() pour attendre de manière bloquante (ou await) l’ID du nouvel
* `OperationRef[str]` : instantané. Appelez `Sandbox.get_snapshot(id)` pour obtenir l’enregistrement complet
* `OperationRef[str]` : (statut, taille, horodatages).
**Exceptions levées**
* `SandboxSnapshotError` : si l’instantané échoue (voir les sous-classes pour `NOT_SUPPORTED` lorsque FSS n’est pas activé pour l’organisation, quota/taille, etc.).
**Exemples**
```python theme={"system"}
with Sandbox.run(
file_system_snapshot=FileSystemSnapshotOptions(mount_path="/workspace"),
) as sb:
sb.exec(["sh", "-c", "echo hi > /workspace/note.txt"]).result()
snapshot_id = sb.snapshot().result()
# Consulter l’enregistrement complet si nécessaire :
snap = Sandbox.get_snapshot(snapshot_id).result()
```
exec
```python theme={"system"}
exec(command: Sequence[str], *, cwd: str | None = None, check: bool = False, timeout_seconds: float | None = None, stdin: bool = False, container: str | None = None) -> Process
```
Exécute une commande et renvoie immédiatement un Process.
Remarque : si le sandbox n’est pas encore à l’état RUNNING, cette méthode attend d’abord qu’il le soit.
Le paramètre timeout\_seconds s’applique uniquement à l’exécution de la commande, et non à
l’attente initiale du statut RUNNING.
**Paramètres**
* `command` (`Sequence[str]`) : commande et arguments à exécuter
* `cwd` (`str | None`) : répertoire de travail dans lequel exécuter la commande. Doit être un chemin absolu. S’il est spécifié, la commande est encapsulée dans un cd du shell.
* `check` (`bool`) : si True, lève SandboxExecutionError lorsque le code de retour est non nul
* `timeout_seconds` (`float | None`) : délai d’expiration de l’exécution de la commande (une fois le sandbox à l’état RUNNING). N’inclut pas le temps d’attente nécessaire pour que le sandbox atteigne le statut RUNNING.
* `stdin` (`bool`) : si True, active la diffusion en continu de stdin. Process.stdin est alors un StreamWriter capable d’envoyer des entrées à la commande. Si False (par défaut), stdin est fermé immédiatement et Process.stdin vaut None.
* `container` (`str | None`) : nom du conteneur dans lequel exécuter la commande. Une valeur vide ou None cible le conteneur principal.
**Renvoie**
* `Process` : handle de processus avec diffusion en continu de stdout/stderr. Appelez .result() pour bloquer
* `Process` : jusqu’à l’obtention du ProcessResult final, ou itérez sur .stdout/.stderr pour obtenir
* `Process` : la sortie en temps réel. Lorsque stdin=True, Process.stdin est un StreamWriter.
**Exceptions levées**
* `ValueError` : si la commande est vide ou si cwd est invalide (chemin vide ou relatif)
**Exemples**
```python theme={"system"}
# Obtenir le résultat directement
process = sb.exec(["echo", "hello"])
result = process.result()
print(result.stdout)
# Avec un répertoire de travail
result = sb.exec(["ls", "-la"], cwd="/app").result()
# Diffuser la sortie en continu, en temps réel
process = sb.exec(["python", "script.py"])
for line in process.stdout:
print(line)
result = process.result()
# Avec envoi de stdin en continu
process = sb.exec(["cat"], stdin=True)
process.stdin.write(b"hello world").result()
process.stdin.close().result()
result = process.result()
# Utilisation asynchrone
result = await sb.exec(["echo", "hello"])
```
shell
```python theme={"system"}
shell(command: Sequence[str] | None = None, *, width: int | None = None, height: int | None = None, container: str | None = None) -> TerminalSession
```
Démarre une session TTY interactive dans la sandbox.
Renvoie une TerminalSession optimisée pour une utilisation interactive du terminal :
sortie en octets bruts (sans décodage ni réencodage), aucune mise en mémoire tampon de la sortie et
stdin en mode fire-and-forget.
**Paramètres**
* `command` (`Sequence[str] | None`) : commande shell à exécuter. Valeur par défaut : \["/bin/bash"]. Accepte une séquence telle que \["/bin/sh"] ou \["/usr/bin/python3"].
* `width` (`int | None`) : largeur initiale du terminal, en colonnes.
* `height` (`int | None`) : hauteur initiale du terminal, en lignes.
* `container` (`str | None`) : nom du conteneur de la session TTY. Si la valeur est vide ou None, le conteneur principal est ciblé.
**Renvoie**
* `TerminalSession` : handle TerminalSession exposant .output (StreamReader\[bytes]),
* `TerminalSession` : .stdin (StreamWriter) et .resize(w, h).
**Exceptions levées**
* `ValueError` : si la commande est explicitement vide.
**Exemple**
```python theme={"system"}
session = sandbox.shell(width=80, height=24)
session.stdin.writeline("echo hello").result()
for chunk in session.output:
sys.stdout.buffer.write(chunk)
exit_code = session.wait()
```
read\_file
```python theme={"system"}
read_file(filepath: str, *, timeout_seconds: float | None = None, container: str | None = None) -> OperationRef[bytes]
```
Lit un fichier depuis la sandbox et renvoie immédiatement un OperationRef.
**Paramètres**
* `filepath` (`str`) : chemin du fichier dans la sandbox
* `timeout_seconds` (`float | None`) : délai d’expiration de l’opération
* `container` (`str | None`) : conteneur dans lequel lire. Une valeur vide ou None cible le conteneur principal.
**Renvoie**
* `OperationRef[bytes]` : OperationRef\[bytes] : utilisez .result() pour bloquer et récupérer le contenu.
**Comportement :** les fichiers jusqu’à \~32 Mio sont lus en un seul appel unaire. Pour les fichiers plus volumineux (jusqu’à \~256 Mio), la lecture bascule de manière transparente en diffusion en continu : le premier basculement de ce type pour une Sandbox donnée est journalisé une seule fois au niveau INFO. Lorsque le serveur indique la taille du fichier, les fichiers de plus de \~256 Mio sont refusés avec `CWSANDBOX_FILE_TOO_LARGE` ; utilisez alors `read_file_streaming`. Quel que soit le mode de lecture, l’intégralité du résultat est conservée en mémoire. Le client ne peut pas toujours connaître à l’avance la taille du fichier distant (par exemple lorsque le backend signale la lecture trop volumineuse par un épuisement des ressources plutôt que par un `CWSANDBOX_FILE_TOO_LARGE` précisant la taille) ; un fichier très volumineux peut donc malgré tout être placé intégralement en mémoire tampon au lieu d’être refusé. Pour tout fichier volumineux, privilégiez `read_file_streaming` afin de le consommer de manière incrémentielle et de limiter l’utilisation de la mémoire.
**Exceptions levées**
* `SandboxFileError` : avec `reason == CWSANDBOX_FILE_TOO_LARGE` lorsque le fichier dépasse le plafond du serveur et que celui-ci a indiqué sa taille ; ou avec `reason == CWSANDBOX_FILE_TRUNCATED` lorsqu’une lecture en diffusion en continu renvoie moins de données que la taille du fichier (troncature détectée par rapport à la taille obtenue avant la lecture).
* `SandboxStreamBackpressureError` : lorsqu’une lecture volumineuse bascule en diffusion en continu et que la sortie est produite plus vite que le client ne la lit (sous-classe de SandboxExecutionError).
**Exemples**
```python theme={"system"}
data = sb.read_file("/output/result.txt").result()
```
write\_file
```python theme={"system"}
write_file(filepath: str, contents: bytes, *, timeout_seconds: float | None = None, container: str | None = None) -> OperationRef[None]
```
Écrit un fichier dans la sandbox et renvoie immédiatement un OperationRef.
**Paramètres**
* `filepath` (`str`) : chemin du fichier dans la sandbox
* `contents` (`bytes`) : contenu du fichier sous forme d’octets
* `timeout_seconds` (`float | None`) : délai d’expiration de l’opération
* `container` (`str | None`) : conteneur dans lequel écrire. Une valeur vide ou None cible le conteneur principal.
**Renvoie**
* `OperationRef[None]` : OperationRef\[None] : utilisez .result() pour bloquer jusqu’à la fin de l’opération.
**Comportement :** les charges utiles allant jusqu’à environ 32 Mio sont écrites en un seul appel unaire. Les charges utiles plus volumineuses (jusqu’à environ 256 Mio) basculent de manière transparente vers une écriture en diffusion en continu : le premier basculement de ce type pour une Sandbox donnée est journalisé une seule fois au niveau INFO. Les charges utiles dépassant environ 256 Mio sont refusées ; utilisez alors `write_file_streaming`.
**Exceptions levées**
* `SandboxFileError` : avec `reason == CWSANDBOX_FILE_TOO_LARGE` lorsque la charge utile dépasse le plafond du serveur, ou (sans ce motif) si une écriture en diffusion en continu échoue en cours de route et a pu laisser un fichier partiel.
* `SandboxStreamBackpressureError` : lorsqu’une écriture volumineuse bascule vers la diffusion en continu et que la source produit des données plus vite qu’elles ne peuvent être envoyées (sous-classe de SandboxExecutionError).
**Exemples**
```python theme={"system"}
sb.write_file("/input/data.txt", b"content").result()
```
write\_file\_streaming
```python theme={"system"}
write_file_streaming(filepath: str, source: bytes | Iterable[bytes] | AsyncIterable[bytes], *, timeout_seconds: float | None = None, container: str | None = None) -> OperationRef[None]
```
Diffuse un fichier vers la sandbox sans matérialiser l’intégralité de la charge utile.
Privilégiez cette méthode à `write_file` pour les charges utiles de plus de
32 Mio environ, ou dès que les données se présentent déjà sous forme d’itérateur (descripteur de fichier,
générateur, producteur asynchrone).
**Paramètres**
* `filepath` (`str`) : chemin absolu dans la sandbox.
* `source` (`bytes | Iterable[bytes] | AsyncIterable[bytes]`) : charge utile sous forme de `bytes`, d’un `Iterable[bytes]` synchrone ou d’un `AsyncIterable[bytes]`. L’entrée est découpée en fragments compatibles avec le tramage avant la transmission. Les éléments produits doivent être de type `bytes`, `bytearray` ou `memoryview` ; tout autre type lève une `TypeError`.
* `timeout_seconds` (`float | None`) : délai d’expiration, en temps réel, de l’écriture en diffusion en continu.
**Renvoie**
* `OperationRef[None]` : `OperationRef[None]` : appelez `.result()` pour bloquer jusqu’à la fin de l’opération.
**Exceptions levées**
* `SandboxStreamBackpressureError` : si la source produit des données plus vite qu’elles ne peuvent être envoyées et que le flux est interrompu prématurément. Utilisez une source dont vous pouvez réguler le débit, ou découpez au préalable les téléversements volumineux en fragments ; consultez la docstring de cette exception pour plus de détails.
**Mises en garde :** la destination est écrite directement (sans fichier temporaire suivi d’un renommage). Une annulation ou une erreur de transport en cours de flux peut laisser un fichier incomplet. Par ailleurs, le transfert en diffusion en continu ne résiste pas à un redémarrage de la sandbox. Une source synchrone (par exemple un descripteur de fichier obtenu avec `open(...)`) est lue sur un thread worker, si bien qu’un `read` bloquant ne fige pas la boucle d’événements du SDK. Une source asynchrone est attendue directement. Les deux conviennent : choisissez celle qui se prête le mieux à vos données.
read\_file\_streaming
```python theme={"system"}
read_file_streaming(filepath: str, *, timeout_seconds: float | None = None, container: str | None = None) -> StreamReader[bytes]
```
Diffuse en continu un fichier depuis la sandbox par fragments, sans placer toute la charge utile en mémoire tampon.
Préférez cette méthode à `read_file` pour les fichiers de plus de 32 Mio environ,
ou chaque fois que vous souhaitez consommer le fichier de manière incrémentielle (écriture sur
disque, calcul d’un hachage à la volée, analyse ligne par ligne).
Pour les fichiers volumineux, le SDK relève la taille du fichier *avant* la lecture et,
une fois le flux terminé, vérifie qu’au moins ce nombre d’octets a bien été
reçu. Si ce n’est pas le cas, l’itérateur lève `SandboxFileError`
avec le motif `CWSANDBOX_FILE_TRUNCATED`, ce qui permet à l’appelant de détecter une lecture
incomplète silencieuse au lieu de consommer un fichier partiel. (Comme la taille est
mesurée avant la lecture, un fichier auquel des données sont ajoutées pendant la lecture n’est jamais pris pour une
troncature.) Cette vérification n’est pas effectuée pour les petits fichiers, qui ne peuvent pas subir de troncature
silencieuse, et elle n’est assurée qu’au mieux lorsque la taille ne peut pas être
déterminée.
Si votre boucle lit les fragments plus lentement que le fichier n’est diffusé (par exemple, si vous effectuez
un traitement lent entre les itérations), la lecture peut être interrompue prématurément par
`SandboxStreamBackpressureError`. Itérez sans attendre et sortez les traitements lents
de la boucle de lecture ; consultez la docstring de cette exception pour plus de détails.
**Paramètres**
* `filepath` (`str`) : chemin absolu dans la sandbox.
* `timeout_seconds` (`float | None`) : délai d’expiration, en temps réel, de la lecture en diffusion en continu.
**Renvoie**
* `StreamReader[bytes]` : `StreamReader[bytes]` produisant les fragments dans l’ordre. La fin du fichier est
* `StreamReader[bytes]` : signalée par l’épuisement normal de l’itérateur. Les erreurs (fichier introuvable,
* `StreamReader[bytes]` : autorisation refusée, troncature, lecteur trop lent) sont levées à nouveau
* `StreamReader[bytes]` : lorsque le consommateur les atteint au cours de l’itération.
**Exemple**
```python theme={"system"}
with contextlib.closing(sb.read_file_streaming("/data/big.bin")) as reader:
with open("local.bin", "wb") as f:
for chunk in reader:
f.write(chunk)
```
**Mises en garde :** le transfert en diffusion en continu ne survit pas à un redémarrage de la sandbox ; un long transfert qui coïncide avec un redémarrage échouera en cours de flux. Les appelants doivent parcourir le lecteur jusqu’à son achèvement ou appeler `close()` sur celui-ci. Le SDK installe un finaliseur qui annule la tâche en arrière-plan lors du garbage collection, mais une fermeture explicite libère les ressources plus tôt. Une quantité limitée de sortie est placée en mémoire tampon en amont de votre boucle afin d’absorber les pics et d’appliquer une contre-pression, mais il ne s’agit pas d’un plafond de mémoire strict : la mémoire résidente continue d’augmenter à mesure que votre boucle prend du retard. Gardez la boucle de lecture aussi légère que possible et déportez hors de celle-ci tout traitement lent par fragment (voir `examples/large_file_streaming.py`).
stream\_logs
```python theme={"system"}
stream_logs(*, follow: bool = False, tail_lines: int | None = None, since_time: datetime | None = None, timestamps: bool = False, timeout_seconds: float | None = None, container: str | None = None) -> StreamReader[str]
```
Diffuse en continu les logs du processus principal de la sandbox.
Diffuse en continu stdout/stderr de la **commande principale** de la sandbox : le
point d’entrée transmis à `Sandbox.run()` (ou, par défaut, la commande de maintien en vie
interceptée par le shell). La sortie des commandes lancées via `exec()` n’est **pas**
incluse ; pour celles-ci, utilisez `Process.stdout`/`Process.stderr`.
.. note::
Les sandboxes créées avec la commande de maintien en vie par défaut ne
produisent aucun log. Pour afficher des logs ici, transmettez une commande qui
écrit sur stdout/stderr lors de l’appel à `Sandbox.run()`.
Renvoie un StreamReader qui génère les lignes de log sous forme de chaînes de caractères. La méthode
rend la main immédiatement ; l’itération sur le StreamReader reste bloquée jusqu’à
l’arrivée de données.
**Paramètres**
* `follow` (`bool`) : si True, diffuse en continu les nouveaux logs (comme `tail -f`). Si False, diffuse les logs existants de la sandbox en cours d’exécution, puis s’arrête. Les sandboxes arrêtées rejettent `StreamLogs`.
* `tail_lines` (`int | None`) : nombre de lignes les plus récentes à récupérer. Si None, renvoie toutes les lignes disponibles.
* `since_time` (`datetime | None`) : renvoie uniquement les logs postérieurs à cet horodatage. Doit inclure un fuseau horaire ; les datetimes naïfs lèvent `ValueError`.
* `timestamps` (`bool`) : si True, préfixe chaque ligne d’un horodatage ISO 8601 fourni par le serveur.
* `timeout_seconds` (`float | None`) : délai limite côté client pour l’appel gRPC. Vaut par défaut `request_timeout_seconds` lorsque `follow=False`, et `None` (aucun délai d’expiration) lorsque `follow=True`.
* `container` (`str | None`) : conteneur dont les logs doivent être diffusés. Une valeur vide ou None cible le conteneur principal.
**Renvoie**
* `StreamReader[str]` : StreamReader générant les lignes de log sous forme de chaînes de caractères. Itérez de manière synchrone
* `StreamReader[str]` : avec `for line in reader` ou de manière asynchrone avec
* `StreamReader[str]` : `async for line in reader`.
**Exceptions levées**
* `SandboxNotRunningError` : si `follow=True` et que la sandbox a été arrêtée.
* `SandboxError` : si le flux de logs rencontre une erreur.
**Exemple**
```python theme={"system"}
# Appel unique : récupérer les journaux récents
for line in sandbox.stream_logs(tail_lines=100):
print(line, end="")
# Mode suivi : diffuser en continu
for line in sandbox.stream_logs(follow=True):
print(line, end="")
# Utilisation asynchrone
async for line in sandbox.stream_logs(follow=True):
print(line, end="")
```