Add functions admin CRUD methods

Author: jwfing

insforge/functions/client.py

@@ -5,14 +5,80 @@
 
 from .._base_client import BaseClient
 from .._utils import quote_path_segment
+from .models import FunctionCreateRequest
 from .models import FunctionDetails
+from .models import FunctionDeleteResponse
+from .models import FunctionMutationResponse
 from .models import FunctionMetadata
+from .models import FunctionUpdateRequest
 
 
 class FunctionsClient:
     def __init__(self, client: BaseClient) -> None:
         self._client = client
 
+    async def create_function(
+        self,
+        *,
+        name: str,
+        slug: str | None = None,
+        code: str,
+        description: str | None = None,
+        status: str | None = None,
+        access_token: str | None = None,
+    ) -> FunctionMutationResponse:
+        payload = FunctionCreateRequest(
+            name=name,
+            slug=slug,
+            code=code,
+            description=description,
+            status=status,
+        ).model_dump(exclude_none=True)
+        response = await self._client._request_json(
+            "POST",
+            "/api/functions",
+            json=payload,
+            access_token=access_token,
+        )
+        return FunctionMutationResponse.model_validate(response)
+
+    async def update_function(
+        self,
+        slug: str,
+        *,
+        name: str | None = None,
+        code: str | None = None,
+        description: str | None = None,
+        status: str | None = None,
+        access_token: str | None = None,
+    ) -> FunctionMutationResponse:
+        payload = FunctionUpdateRequest(
+            name=name,
+            code=code,
+            description=description,
+            status=status,
+        ).model_dump(exclude_none=True)
+        response = await self._client._request_json(
+            "PUT",
+            f"/api/functions/{quote_path_segment(slug)}",
+            json=payload,
+            access_token=access_token,
+        )
+        return FunctionMutationResponse.model_validate(response)
+
+    async def delete_function(
+        self,
+        slug: str,
+        *,
+        access_token: str | None = None,
+    ) -> FunctionDeleteResponse:
+        response = await self._client._request_json(
+            "DELETE",
+            f"/api/functions/{quote_path_segment(slug)}",
+            access_token=access_token,
+        )
+        return FunctionDeleteResponse.model_validate(response)
+
     async def list_functions(self, *, access_token: str | None = None) -> list[FunctionMetadata]:
         payload = await self._client._request_json(
             "GET",

insforge/functions/models.py

@@ -21,3 +21,36 @@ class FunctionMetadata(BaseModel):
 
 class FunctionDetails(FunctionMetadata):
     code: str
+
+
+class FunctionCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    name: str
+    slug: str | None = None
+    code: str
+    description: str | None = None
+    status: str | None = None
+
+
+class FunctionUpdateRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    name: str | None = None
+    code: str | None = None
+    description: str | None = None
+    status: str | None = None
+
+
+class FunctionMutationResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    success: bool
+    function: FunctionMetadata
+
+
+class FunctionDeleteResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    success: bool
+    message: str | None = None

tests/functions/test_functions_client.py

@@ -3,7 +3,9 @@
 import httpx
 
 from insforge import InsforgeClient
+from insforge.functions.models import FunctionDeleteResponse
 from insforge.functions.models import FunctionDetails
+from insforge.functions.models import FunctionMutationResponse
 from insforge.functions.models import FunctionMetadata
 
 
@@ -208,3 +210,158 @@ async def fake_request(method: str, url: httpx.URL, **kwargs: object) -> httpx.R
     assert isinstance(result, FunctionDetails)
     assert result.slug == "hello-world"
     assert result.code.startswith("export default")
+
+
+def test_create_function_uses_api_path_and_returns_typed_mutation_response() -> None:
+    async def scenario() -> tuple[object, dict[str, object]]:
+        captured: dict[str, object] = {}
+
+        async def fake_request(method: str, url: httpx.URL, **kwargs: object) -> httpx.Response:
+            captured["method"] = method
+            captured["url"] = str(url)
+            captured["kwargs"] = kwargs
+            return httpx.Response(
+                201,
+                json={
+                    "success": True,
+                    "function": {
+                        "id": "123e4567-e89b-12d3-a456-426614174000",
+                        "slug": "hello-world",
+                        "name": "Hello World Function",
+                        "description": "Returns a greeting message",
+                        "status": "active",
+                        "created_at": "2024-01-21T10:30:00Z",
+                        "updated_at": "2024-01-21T10:35:00Z",
+                        "deployed_at": "2024-01-21T10:35:00Z",
+                    },
+                },
+            )
+
+        async with InsforgeClient(
+            base_url="https://example.com",
+            api_key="ins_test",
+        ) as client:
+            client.http_client.request = fake_request  # type: ignore[method-assign]
+            result = await client.functions.create_function(
+                name="Hello World Function",
+                slug="hello-world",
+                code="export default async function () { return new Response('hi'); }",
+                description="Returns a greeting message",
+                status="active",
+                access_token="admin_token",
+            )
+            return result, captured
+
+    result, captured = asyncio.run(scenario())
+
+    assert captured["method"] == "POST"
+    assert captured["url"] == "https://example.com/api/functions"
+    assert captured["kwargs"]["headers"]["X-API-Key"] == "ins_test"
+    assert captured["kwargs"]["headers"]["Authorization"] == "Bearer admin_token"
+    assert captured["kwargs"]["json"] == {
+        "name": "Hello World Function",
+        "slug": "hello-world",
+        "code": "export default async function () { return new Response('hi'); }",
+        "description": "Returns a greeting message",
+        "status": "active",
+    }
+    assert isinstance(result, FunctionMutationResponse)
+    assert result.success is True
+    assert isinstance(result.function, FunctionMetadata)
+    assert result.function.slug == "hello-world"
+
+
+def test_update_function_uses_api_path_and_returns_typed_mutation_response() -> None:
+    async def scenario() -> tuple[object, dict[str, object]]:
+        captured: dict[str, object] = {}
+
+        async def fake_request(method: str, url: httpx.URL, **kwargs: object) -> httpx.Response:
+            captured["method"] = method
+            captured["url"] = str(url)
+            captured["kwargs"] = kwargs
+            return httpx.Response(
+                200,
+                json={
+                    "success": True,
+                    "function": {
+                        "id": "123e4567-e89b-12d3-a456-426614174000",
+                        "slug": "hello-world",
+                        "name": "Hello World Function v2",
+                        "description": "Returns a greeting message",
+                        "status": "active",
+                        "created_at": "2024-01-21T10:30:00Z",
+                        "updated_at": "2024-01-21T11:00:00Z",
+                        "deployed_at": "2024-01-21T10:35:00Z",
+                    },
+                },
+            )
+
+        async with InsforgeClient(
+            base_url="https://example.com",
+            api_key="ins_test",
+        ) as client:
+            client.http_client.request = fake_request  # type: ignore[method-assign]
+            result = await client.functions.update_function(
+                "hello-world",
+                name="Hello World Function v2",
+                code="export default async function () { return new Response('hi'); }",
+                description="Returns a greeting message",
+                status="active",
+                access_token="admin_token",
+            )
+            return result, captured
+
+    result, captured = asyncio.run(scenario())
+
+    assert captured["method"] == "PUT"
+    assert captured["url"] == "https://example.com/api/functions/hello-world"
+    assert captured["kwargs"]["headers"]["X-API-Key"] == "ins_test"
+    assert captured["kwargs"]["headers"]["Authorization"] == "Bearer admin_token"
+    assert captured["kwargs"]["json"] == {
+        "name": "Hello World Function v2",
+        "code": "export default async function () { return new Response('hi'); }",
+        "description": "Returns a greeting message",
+        "status": "active",
+    }
+    assert isinstance(result, FunctionMutationResponse)
+    assert result.success is True
+    assert result.function.name == "Hello World Function v2"
+
+
+def test_delete_function_uses_api_path_and_returns_typed_delete_response() -> None:
+    async def scenario() -> tuple[object, dict[str, object]]:
+        captured: dict[str, object] = {}
+
+        async def fake_request(method: str, url: httpx.URL, **kwargs: object) -> httpx.Response:
+            captured["method"] = method
+            captured["url"] = str(url)
+            captured["kwargs"] = kwargs
+            return httpx.Response(
+                200,
+                json={
+                    "success": True,
+                    "message": "Function hello-world deleted successfully",
+                },
+            )
+
+        async with InsforgeClient(
+            base_url="https://example.com",
+            api_key="ins_test",
+        ) as client:
+            client.http_client.request = fake_request  # type: ignore[method-assign]
+            result = await client.functions.delete_function(
+                "hello-world",
+                access_token="admin_token",
+            )
+            return result, captured
+
+    result, captured = asyncio.run(scenario())
+
+    assert captured["method"] == "DELETE"
+    assert captured["url"] == "https://example.com/api/functions/hello-world"
+    assert captured["kwargs"]["headers"]["X-API-Key"] == "ins_test"
+    assert captured["kwargs"]["headers"]["Authorization"] == "Bearer admin_token"
+    assert captured["kwargs"].get("json") is None
+    assert isinstance(result, FunctionDeleteResponse)
+    assert result.success is True
+    assert result.message == "Function hello-world deleted successfully"