feat: sdk.revisions.deploy

src/deno_sandbox/__init__.py

@@ -14,7 +14,16 @@
     RuntimeLog,
     RuntimeLogsResponse,
 )
-from .revisions import Revisions, AsyncRevisions, Revision, RevisionListItem
+from .revisions import (
+    Revisions,
+    AsyncRevisions,
+    Revision,
+    RevisionListItem,
+    FileAsset,
+    SymlinkAsset,
+    Asset,
+    EnvVarInputForDeploy,
+)
 from .snapshots import Snapshots, AsyncSnapshots
 from .timelines import Timelines, AsyncTimelines
 from .volumes import Volumes, AsyncVolumes
@@ -42,6 +51,10 @@
     "RuntimeLogsResponse",
     "Revision",
     "RevisionListItem",
+    "FileAsset",
+    "SymlinkAsset",
+    "Asset",
+    "EnvVarInputForDeploy",
 ]
 
 

src/deno_sandbox/revisions.py

@@ -1,11 +1,30 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any, TypedDict, cast, overload
+from typing import Any, Dict, TypedDict, Union, cast, overload
 from typing_extensions import Literal, NotRequired, Optional
 
 from deno_sandbox.apps import Config, EnvVar, LayerRef
 
+
+class FileAsset(TypedDict):
+    kind: Literal["file"]
+    encoding: Literal["utf-8", "base64"]
+    content: str
+
+
+class SymlinkAsset(TypedDict):
+    kind: Literal["symlink"]
+    target: str
+
+
+Asset = Union[FileAsset, SymlinkAsset]
+
+
+class EnvVarInputForDeploy(TypedDict):
+    key: str
+    value: str
+
 from .bridge import AsyncBridge
 from .console import (
     AsyncConsoleClient,
@@ -154,6 +173,37 @@ async def cancel(self, revision: str) -> Revision:
         result = await self._client.post(f"/api/v2/revisions/{revision}/cancel", {})
         return cast(Revision, convert_to_snake_case(result))
 
+    async def create(
+        self,
+        app: str,
+        assets: Dict[str, Asset],
+        *,
+        config: Optional[Config] = None,
+        env_vars: Optional[list[EnvVarInputForDeploy]] = None,
+        labels: Optional[Dict[str, str]] = None,
+    ) -> Revision:
+        """Create a revision by uploading source files as assets.
+
+        Args:
+            app: The app ID or slug.
+            assets: Dict mapping file paths to Asset objects.
+            config: Optional build/runtime configuration.
+            env_vars: Optional environment variables for this revision.
+            labels: Optional labels (e.g., git metadata).
+
+        Returns:
+            The created Revision (build is async; poll for status).
+        """
+        body: Dict[str, Any] = {"assets": assets}
+        if config is not None:
+            body["config"] = config
+        if env_vars is not None:
+            body["env_vars"] = env_vars
+        if labels is not None:
+            body["labels"] = labels
+        result = await self._client.post(f"/api/v2/apps/{app}/deploy", body)
+        return cast(Revision, convert_to_snake_case(result))
+
 
 class Revisions:
     def __init__(self, client: AsyncConsoleClient, bridge: AsyncBridge):
@@ -203,3 +253,30 @@ def cancel(self, revision: str) -> Revision:
             revision: The revision ID.
         """
         return self._bridge.run(self._async.cancel(revision))
+
+    def create(
+        self,
+        app: str,
+        assets: Dict[str, Asset],
+        *,
+        config: Optional[Config] = None,
+        env_vars: Optional[list[EnvVarInputForDeploy]] = None,
+        labels: Optional[Dict[str, str]] = None,
+    ) -> Revision:
+        """Create a revision by uploading source files as assets.
+
+        Args:
+            app: The app ID or slug.
+            assets: Dict mapping file paths to Asset objects.
+            config: Optional build/runtime configuration.
+            env_vars: Optional environment variables for this revision.
+            labels: Optional labels (e.g., git metadata).
+
+        Returns:
+            The created Revision (build is async; poll for status).
+        """
+        return self._bridge.run(
+            self._async.create(
+                app, assets, config=config, env_vars=env_vars, labels=labels
+            )
+        )

tests/test_revisions.py

@@ -139,3 +139,56 @@ async def test_revisions_get_deprecated_two_arg_async():
         assert fetched["id"] == revision["id"]
     finally:
         await sdk.apps.delete(app["id"])
+
+
+@pytest.mark.timeout(60)
+@pytest.mark.asyncio(loop_scope="session")
+async def test_revisions_create_async():
+    sdk = AsyncDenoDeploy()
+    app = await sdk.apps.create()
+    try:
+        revision = await sdk.revisions.create(
+            app["id"],
+            assets={
+                "main.ts": {
+                    "kind": "file",
+                    "encoding": "utf-8",
+                    "content": 'Deno.serve(() => new Response("Hello"))',
+                }
+            },
+        )
+        assert revision["id"] is not None
+        assert revision["status"] in [
+            "queued",
+            "building",
+            "succeeded",
+            "failed",
+        ]
+    finally:
+        await sdk.apps.delete(app["id"])
+
+
+@pytest.mark.timeout(60)
+def test_revisions_create_sync():
+    sdk = DenoDeploy()
+    app = sdk.apps.create()
+    try:
+        revision = sdk.revisions.create(
+            app["id"],
+            assets={
+                "main.ts": {
+                    "kind": "file",
+                    "encoding": "utf-8",
+                    "content": 'Deno.serve(() => new Response("Hello"))',
+                }
+            },
+        )
+        assert revision["id"] is not None
+        assert revision["status"] in [
+            "queued",
+            "building",
+            "succeeded",
+            "failed",
+        ]
+    finally:
+        sdk.apps.delete(app["id"])