feat: layers

src/deno_sandbox/__init__.py

@@ -24,6 +24,12 @@
     Asset,
     EnvVarInputForDeploy,
 )
+from .layers import (
+    Layers,
+    AsyncLayers,
+    Layer,
+    LayerAppRef,
+)
 from .snapshots import Snapshots, AsyncSnapshots
 from .timelines import Timelines, AsyncTimelines
 from .volumes import Volumes, AsyncVolumes
@@ -55,6 +61,8 @@
     "SymlinkAsset",
     "Asset",
     "EnvVarInputForDeploy",
+    "Layer",
+    "LayerAppRef",
 ]
 
 
@@ -66,6 +74,7 @@ def __init__(self, options: Optional[Options] = None):
         client = AsyncConsoleClient(internal_options)
         self.apps = Apps(client, bridge)
         self.revisions = Revisions(client, bridge)
+        self.layers = Layers(client, bridge)
         self.timelines = Timelines(client, bridge)
         self.sandbox = SandboxApi(client, bridge)
         self.snapshots = Snapshots(client, bridge)
@@ -79,6 +88,7 @@ def __init__(self, options: Optional[Options] = None):
 
         self.apps = AsyncApps(client)
         self.revisions = AsyncRevisions(client)
+        self.layers = AsyncLayers(client)
         self.timelines = AsyncTimelines(client)
         self.sandbox = AsyncSandboxApi(client)
         self.snapshots = AsyncSnapshots(client)

src/deno_sandbox/layers.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+from typing import Any, List, TypedDict, cast
+from typing_extensions import NotRequired, Optional
+
+from .apps import EnvVar, EnvVarInput, EnvVarUpdate, LayerRef
+from .bridge import AsyncBridge
+from .console import (
+    AsyncConsoleClient,
+    AsyncPaginatedList,
+    PaginatedList,
+)
+from .utils import convert_to_snake_case
+
+
+class Layer(TypedDict):
+    id: str
+    """Unique layer identifier."""
+
+    slug: str
+    """Human-readable layer slug."""
+
+    description: NotRequired[str]
+    """Optional description of the layer's purpose."""
+
+    layers: List[LayerRef]
+    """Base layers included by this layer, in priority order."""
+
+    env_vars: List[EnvVar]
+    """Environment variables defined in this layer."""
+
+    app_count: int
+    """Number of apps that reference this layer."""
+
+    created_at: str
+    """ISO 8601 timestamp of creation."""
+
+    updated_at: str
+    """ISO 8601 timestamp of last modification."""
+
+
+class LayerAppRef(TypedDict):
+    id: str
+    """Unique app identifier (UUID)."""
+
+    slug: str
+    """Human-readable app slug."""
+
+    layer_position: int
+    """Index of this layer in the app's layers array."""
+
+
+class AsyncLayers:
+    def __init__(self, client: AsyncConsoleClient):
+        self._client = client
+
+    async def get(self, id_or_slug: str) -> Layer | None:
+        """Get a layer by ID or slug.
+
+        Args:
+            id_or_slug: Layer ID or slug.
+        """
+        result = await self._client.get_or_none(f"/api/v2/layers/{id_or_slug}")
+        if result is None:
+            return None
+        return cast(Layer, convert_to_snake_case(result))
+
+    async def list(
+        self,
+        *,
+        search: Optional[str] = None,
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> AsyncPaginatedList[Layer]:
+        """List all layers in the organization.
+
+        Args:
+            search: Search query for filtering.
+            cursor: Pagination cursor.
+            limit: Maximum number of items to return.
+        """
+        params: dict[str, Any] = {}
+        if search is not None:
+            params["search"] = search
+        if cursor is not None:
+            params["cursor"] = cursor
+        if limit is not None:
+            params["limit"] = limit
+        return await self._client.get_paginated(
+            "/api/v2/layers",
+            cursor=None,
+            params=params if params else None,
+        )
+
+    async def create(
+        self,
+        slug: str,
+        *,
+        description: Optional[str] = None,
+        layers: Optional[List[str]] = None,
+        env_vars: Optional[List[EnvVarInput]] = None,
+    ) -> Layer:
+        """Create a new layer.
+
+        Args:
+            slug: Human-readable layer slug.
+            description: Optional description of the layer's purpose.
+            layers: Base layer IDs or slugs to include.
+            env_vars: Environment variables for this layer.
+        """
+        body: dict[str, Any] = {"slug": slug}
+        if description is not None:
+            body["description"] = description
+        if layers is not None:
+            body["layers"] = layers
+        if env_vars is not None:
+            body["env_vars"] = env_vars
+        result = await self._client.post("/api/v2/layers", body)
+        return cast(Layer, convert_to_snake_case(result))
+
+    async def update(
+        self,
+        layer: str,
+        *,
+        slug: Optional[str] = None,
+        description: Optional[str] = None,
+        layers: Optional[List[str]] = None,
+        env_vars: Optional[List[EnvVarUpdate]] = None,
+    ) -> Layer:
+        """Update a layer.
+
+        Args:
+            layer: Layer ID or slug.
+            slug: New layer slug.
+            description: New description.
+            layers: Replace all included layers.
+            env_vars: Deep merge with existing environment variables.
+        """
+        body: dict[str, Any] = {}
+        if slug is not None:
+            body["slug"] = slug
+        if description is not None:
+            body["description"] = description
+        if layers is not None:
+            body["layers"] = layers
+        if env_vars is not None:
+            body["env_vars"] = env_vars
+        result = await self._client.patch(f"/api/v2/layers/{layer}", body)
+        return cast(Layer, convert_to_snake_case(result))
+
+    async def delete(self, layer: str) -> None:
+        """Delete a layer.
+
+        Args:
+            layer: Layer ID or slug.
+        """
+        await self._client.delete(f"/api/v2/layers/{layer}")
+
+    async def apps(
+        self,
+        layer: str,
+        *,
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> AsyncPaginatedList[LayerAppRef]:
+        """List apps that reference this layer.
+
+        Args:
+            layer: Layer ID or slug.
+            cursor: Pagination cursor.
+            limit: Maximum number of items to return.
+        """
+        params: dict[str, Any] = {}
+        if cursor is not None:
+            params["cursor"] = cursor
+        if limit is not None:
+            params["limit"] = limit
+        return await self._client.get_paginated(
+            f"/api/v2/layers/{layer}/apps",
+            cursor=None,
+            params=params if params else None,
+        )
+
+
+class Layers:
+    def __init__(self, client: AsyncConsoleClient, bridge: AsyncBridge):
+        self._client = client
+        self._bridge = bridge
+        self._async = AsyncLayers(client)
+
+    def get(self, id_or_slug: str) -> Layer | None:
+        """Get a layer by ID or slug.
+
+        Args:
+            id_or_slug: Layer ID or slug.
+        """
+        return self._bridge.run(self._async.get(id_or_slug))
+
+    def list(
+        self,
+        *,
+        search: Optional[str] = None,
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> PaginatedList[Layer]:
+        """List all layers in the organization.
+
+        Args:
+            search: Search query for filtering.
+            cursor: Pagination cursor.
+            limit: Maximum number of items to return.
+        """
+        paginated = self._bridge.run(
+            self._async.list(search=search, cursor=cursor, limit=limit)
+        )
+        return PaginatedList(self._bridge, paginated)
+
+    def create(
+        self,
+        slug: str,
+        *,
+        description: Optional[str] = None,
+        layers: Optional[List[str]] = None,
+        env_vars: Optional[List[EnvVarInput]] = None,
+    ) -> Layer:
+        """Create a new layer.
+
+        Args:
+            slug: Human-readable layer slug.
+            description: Optional description of the layer's purpose.
+            layers: Base layer IDs or slugs to include.
+            env_vars: Environment variables for this layer.
+        """
+        return self._bridge.run(
+            self._async.create(
+                slug, description=description, layers=layers, env_vars=env_vars
+            )
+        )
+
+    def update(
+        self,
+        layer: str,
+        *,
+        slug: Optional[str] = None,
+        description: Optional[str] = None,
+        layers: Optional[List[str]] = None,
+        env_vars: Optional[List[EnvVarUpdate]] = None,
+    ) -> Layer:
+        """Update a layer.
+
+        Args:
+            layer: Layer ID or slug.
+            slug: New layer slug.
+            description: New description.
+            layers: Replace all included layers.
+            env_vars: Deep merge with existing environment variables.
+        """
+        return self._bridge.run(
+            self._async.update(
+                layer,
+                slug=slug,
+                description=description,
+                layers=layers,
+                env_vars=env_vars,
+            )
+        )
+
+    def delete(self, layer: str) -> None:
+        """Delete a layer.
+
+        Args:
+            layer: Layer ID or slug.
+        """
+        self._bridge.run(self._async.delete(layer))
+
+    def apps(
+        self,
+        layer: str,
+        *,
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> PaginatedList[LayerAppRef]:
+        """List apps that reference this layer.
+
+        Args:
+            layer: Layer ID or slug.
+            cursor: Pagination cursor.
+            limit: Maximum number of items to return.
+        """
+        paginated = self._bridge.run(
+            self._async.apps(layer, cursor=cursor, limit=limit)
+        )
+        return PaginatedList(self._bridge, paginated)