sdk-python: break the client/template import cycle; use Client.attach in the openai adapter

sdk/openai/cocoonsandbox_openai/adapter.py

@@ -126,7 +126,7 @@ def _sandbox(self) -> Sandbox:
         if self._sb is None:
             s = self.state
             client = Client(s.addr, api_token=s.api_token)
-            self._sb = Sandbox(client=client, id=s.sandbox_id, token=s.sandbox_token, owner=s.owner or s.addr)
+            self._sb = client.attach(s.owner or s.addr, s.sandbox_id, s.sandbox_token)
         return self._sb
 
     def _abs(self, path: Path | str) -> Path:

sdk/python/cocoonsandbox/client.py

@@ -17,7 +17,7 @@
 from .endpoint import _endpoint_url
 from .errors import APIError, SandboxTimeout, require_field
 from .sandbox import Sandbox
-from .volumes import _volume_body
+from .template import _claim_body, _template_query
 
 T = TypeVar("T")
 
@@ -279,40 +279,6 @@ def _tls(self) -> ssl.SSLContext:
         return self._ssl_context
 
 
-def _claim_body(
-    template: str,
-    net: str,
-    size: str,
-    ttl_seconds: int,
-    volumes: list[str | Mapping[str, str]] | None = None,
-    mount: bool = True,
-    claim_ref: str = "",
-) -> dict[str, Any]:
-    claim: dict[str, Any] = {"template": template}
-    if net:
-        claim["net"] = net
-    if size:
-        claim["size"] = size
-    if ttl_seconds:
-        claim["ttl_seconds"] = ttl_seconds
-    if volumes:
-        claim["volumes"] = [_volume_body(volume, mount) for volume in volumes]
-    if not mount:
-        claim["volumes_attach_only"] = True
-    if claim_ref:
-        claim["claim_ref"] = claim_ref
-    return claim
-
-
-def _template_query(template: str, net: str, size: str) -> dict[str, str]:
-    query = {"template": template}
-    if net:
-        query["net"] = net
-    if size:
-        query["size"] = size
-    return query
-
-
 def _error_message(raw: bytes) -> str:
     try:
         return cast(str, json.loads(raw)["error"])

sdk/python/cocoonsandbox/template.py

@@ -4,7 +4,9 @@
 
 import urllib.parse
 from collections.abc import Mapping
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+
+from .volumes import _volume_body
 
 if TYPE_CHECKING:
     from .client import Client
@@ -32,9 +34,6 @@ def new(
     ) -> Sandbox:
         """Claims the template, following placement when volumes require it.
         mount=False attaches the volumes without mounting them."""
-        # local import: a top-level one closes the client -> sandbox -> template cycle.
-        from .client import _claim_body
-
         claim = _claim_body(self.name, self.net, self.size, ttl_seconds, volumes, mount)
         claim["require_promoted"] = True
         if volumes:
@@ -44,10 +43,42 @@ def new(
         return self._client._handle_from(self._addr, reply)
 
     def delete(self) -> None:
-        from .client import _template_query
-
         query = _template_query(self.name, self.net, self.size)
         query["no_redirect"] = "1"
         self._client._request(
             self._addr, "DELETE", "/v1/templates?" + urllib.parse.urlencode(query), None, "delete template"
         )
+
+
+def _claim_body(
+    template: str,
+    net: str,
+    size: str,
+    ttl_seconds: int,
+    volumes: list[str | Mapping[str, str]] | None = None,
+    mount: bool = True,
+    claim_ref: str = "",
+) -> dict[str, Any]:
+    claim: dict[str, Any] = {"template": template}
+    if net:
+        claim["net"] = net
+    if size:
+        claim["size"] = size
+    if ttl_seconds:
+        claim["ttl_seconds"] = ttl_seconds
+    if volumes:
+        claim["volumes"] = [_volume_body(volume, mount) for volume in volumes]
+    if not mount:
+        claim["volumes_attach_only"] = True
+    if claim_ref:
+        claim["claim_ref"] = claim_ref
+    return claim
+
+
+def _template_query(template: str, net: str, size: str) -> dict[str, str]:
+    query = {"template": template}
+    if net:
+        query["net"] = net
+    if size:
+        query["size"] = size
+    return query