fix(remote): align Function CRUD routes (#4166)

Align the experimental Function HTTP transport with the equivalent Table CRUD API shape. This is an intentional breaking change to the experimental Function routes; public Rust and Python APIs remain unchanged. ## Route comparison | Operation | Function before | Function after | Equivalent Table API | | --- | --- | --- | --- | | Create | `POST /v1/functions/create` | `POST /v1/function/{id}/create` | `POST /v1/table/{id}/create` | | Describe | `POST /v1/functions/describe` | `POST /v1/function/{id}/describe` | `POST /v1/table/{id}/describe` | | List | `POST /v1/functions/list` | `GET /v1/namespace/{id}/function/list` | `GET /v1/namespace/{id}/table/list` | | Drop | `POST /v1/functions/drop` | `POST /v1/function/{id}/drop` | `POST /v1/table/{id}/drop` | ## Contract details - Create, describe, and drop use a singular resource path. Their `{id}` path parameter is the URL-encoded Function name, and the duplicate Function identifier is removed from each request body. - Create continues to accept `202 Accepted`. - List changes from a POST with a JSON body to a namespace-scoped GET. Its `{id}` path parameter is the namespace identifier rather than a Function name. - Functions do not support nested namespaces yet, so the client lists against the root namespace identifier (`$` with the default delimiter). A non-root namespace is rejected. - The optional list filter is named `name`. `limit`, `page_token`, and `include_definition` remain available as query parameters. - The paginated list response shape is unchanged.
2026-09-21 12:35:42 +00:00 · 2026-09-13 21:44:13 -07:00
parent b8f0048b5a
commit 9fe10c7362
2 changed files with 101 additions and 80 deletions
@@ -15,6 +15,7 @@ from pathlib import Path
 import subprocess
 import sys
 import threading
+import urllib.parse
 from typing import Optional

 import pyarrow as pa
@@ -1213,14 +1214,22 @@ def _mock_remote_function_catalog():
        def log_message(self, *args):
            pass

+        def _write_response(self, status, response):
+            encoded = json.dumps(response).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(encoded)))
+            self.end_headers()
+            self.wfile.write(encoded)
+
        def do_POST(self):
            length = int(self.headers.get("Content-Length", "0"))
            body = json.loads(self.rfile.read(length) or b"{}")
            state["requests"].append((self.path, body))
            status = 200
-            if self.path == "/v1/functions/create":
+            if self.path == "/v1/function/normalize_score/create":
                state["version"] = {
-                    "name": body["name"],
+                    "name": "normalize_score",
                    "version": "fv_exact",
                    "artifact": {
                        key: body["artifact"][key]
@@ -1242,43 +1251,43 @@ def _mock_remote_function_catalog():
                    "job_state": "DONE",
                    "result": state["version"],
                }
-            elif self.path == "/v1/functions/describe":
-                assert body == {
-                    "name": "normalize_score",
-                    "version": "fv_exact",
-                }
+            elif self.path == "/v1/function/normalize_score/describe":
+                assert body == {"version": "fv_exact"}
                response = state["version"]
-            elif self.path == "/v1/functions/list":
-                assert body["include_definition"] is True
-                if "page_token" not in body:
-                    response = {
-                        "functions": [
-                            {
-                                "name": "normalize_score",
-                                "version": "fv_exact",
-                                "definition": state["version"],
-                            }
-                        ],
-                        "page_token": "next",
-                    }
-                else:
-                    assert body["page_token"] == "next"
-                    response = {"functions": []}
-            elif self.path == "/v1/functions/drop":
-                assert body == {
-                    "name": "normalize_score",
-                    "version": "fv_exact",
-                }
+            elif self.path == "/v1/function/normalize_score/drop":
+                assert body == {"version": "fv_exact"}
                response = {"dropped": True}
            else:
                status = 404
                response = {"error": "not found"}
-            encoded = json.dumps(response).encode()
-            self.send_response(status)
-            self.send_header("Content-Type", "application/json")
-            self.send_header("Content-Length", str(len(encoded)))
-            self.end_headers()
-            self.wfile.write(encoded)
+            self._write_response(status, response)
+
+        def do_GET(self):
+            url = urllib.parse.urlsplit(self.path)
+            query = {
+                key: values[-1]
+                for key, values in urllib.parse.parse_qs(url.query).items()
+            }
+            state["requests"].append((url.path, query))
+            if url.path != "/v1/namespace/$/function/list":
+                self._write_response(404, {"error": "not found"})
+                return
+            assert query["include_definition"] == "true"
+            if "page_token" not in query:
+                response = {
+                    "functions": [
+                        {
+                            "name": "normalize_score",
+                            "version": "fv_exact",
+                            "definition": state["version"],
+                        }
+                    ],
+                    "page_token": "next",
+                }
+            else:
+                assert query["page_token"] == "next"
+                response = {"functions": []}
+            self._write_response(200, response)

    with http.server.HTTPServer(("localhost", 0), Handler) as server:
        thread = threading.Thread(target=server.serve_forever)
@@ -1307,9 +1316,11 @@ def test_remote_registration_job_and_exact_version_reopen_round_trip():
    assert reopened.name == "normalize_score"
    assert reopened.version == "fv_exact"
    create_request = state["requests"][0][1]
-    assert create_request == json.loads(
+    expected_request = json.loads(
        normalize_score.registration_request.to_canonical_json()
    )
+    expected_request.pop("name")
+    assert create_request == expected_request


 def test_blocking_remote_registration_returns_function_version():
@@ -1325,7 +1336,7 @@ def test_blocking_remote_registration_returns_function_version():
    assert created.name == "normalize_score"
    assert created.version == "fv_exact"
    assert [path for path, _ in state["requests"]] == [
-        "/v1/functions/create",
+        "/v1/function/normalize_score/create",
        "/v1/jobs/describe",
    ]

@@ -1344,10 +1355,10 @@ def test_remote_list_functions_paginates_and_returns_typed_versions():

    assert functions == [created]
    assert state["requests"] == [
-        ("/v1/functions/list", {"include_definition": True}),
+        ("/v1/namespace/$/function/list", {"include_definition": "true"}),
        (
-            "/v1/functions/list",
-            {"include_definition": True, "page_token": "next"},
+            "/v1/namespace/$/function/list",
+            {"include_definition": "true", "page_token": "next"},
        ),
    ]

@@ -1368,8 +1379,8 @@ async def test_async_remote_list_functions_returns_typed_versions():

    assert functions == [created]
    assert [path for path, _ in state["requests"]] == [
-        "/v1/functions/list",
-        "/v1/functions/list",
+        "/v1/namespace/$/function/list",
+        "/v1/namespace/$/function/list",
    ]


@@ -1385,8 +1396,8 @@ def test_remote_drop_function_sends_exact_version():

    assert state["requests"] == [
        (
-            "/v1/functions/drop",
-            {"name": "normalize_score", "version": "fv_exact"},
+            "/v1/function/normalize_score/drop",
+            {"version": "fv_exact"},
        )
    ]

@@ -1404,7 +1415,7 @@ async def test_async_remote_drop_function_sends_exact_version():

    assert state["requests"] == [
        (
-            "/v1/functions/drop",
-            {"name": "normalize_score", "version": "fv_exact"},
+            "/v1/function/normalize_score/drop",
+            {"version": "fv_exact"},
        )
    ]
@@ -591,7 +591,15 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
        &self,
        request: FunctionRegistrationRequest,
    ) -> Result<Job<FunctionVersion>> {
-        let req = self.client.post("/v1/functions/create").json(&request);
+        let function_id = urlencoding::encode(&request.name);
+        let req = self
+            .client
+            .post(&format!("/v1/function/{function_id}/create"))
+            .json(&serde_json::json!({
+                "artifact": request.artifact,
+                "signature": request.signature,
+                "runtime": request.runtime,
+            }));
        let (request_id, response) = self.client.send(req).await?;
        let response = self.client.check_response(&request_id, response).await?;
        let status = response.status();
@@ -608,11 +616,11 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn get_function(&self, name: &str, version: &str) -> Result<FunctionVersion> {
+        let function_id = urlencoding::encode(name);
        let req = self
            .client
-            .post("/v1/functions/describe")
+            .post(&format!("/v1/function/{function_id}/describe"))
            .json(&serde_json::json!({
-                "name": name,
                "version": version,
            }));
        let (request_id, response) = self.client.send(req).await?;
@@ -621,15 +629,19 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn list_functions(&self) -> Result<Vec<FunctionVersion>> {
+        let namespace_id = build_namespace_identifier(&[], &self.client.id_delimiter);
+        let path = format!("/v1/namespace/{namespace_id}/function/list");
        let mut functions = Vec::new();
        let mut page_token: Option<String> = None;
        let mut seen_page_tokens = HashSet::new();
        loop {
-            let mut body = serde_json::json!({ "include_definition": true });
+            let mut req = self
+                .client
+                .get(&path)
+                .query(&[("include_definition", true)]);
            if let Some(token) = &page_token {
-                body["page_token"] = serde_json::Value::String(token.clone());
+                req = req.query(&[("page_token", token)]);
            }
-            let req = self.client.post("/v1/functions/list").json(&body);
            let (request_id, response) = self.client.send(req).await?;
            let response = self.client.check_response(&request_id, response).await?;
            let status = response.status();
@@ -658,11 +670,11 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn drop_function(&self, name: &str, version: &str) -> Result<bool> {
+        let function_id = urlencoding::encode(name);
        let req = self
            .client
-            .post("/v1/functions/drop")
+            .post(&format!("/v1/function/{function_id}/drop"))
            .json(&serde_json::json!({
-                "name": name,
                "version": version,
            }));
        let (request_id, response) = self.client.send(req).await?;
@@ -2788,9 +2800,10 @@ mod tests {
        );
        const FUNCTION_JOB: &str =
            include_str!("../../tests/fixtures/first_class_functions/v1/remote_function_job.json");
-        let expected: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
+        let mut expected: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
+        expected.as_object_mut().unwrap().remove("name");
        let conn = Connection::new_with_handler(move |request| match request.url().path() {
-            "/v1/functions/create" => {
+            "/v1/function/normalize_score/create" => {
                assert_eq!(request.method(), &reqwest::Method::POST);
                let body: serde_json::Value =
                    serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
@@ -2821,13 +2834,10 @@ mod tests {
        );
        let conn = Connection::new_with_handler(|request| {
            assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/describe");
+            assert_eq!(request.url().path(), "/v1/function/embed/describe");
            let body: serde_json::Value =
                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(
-                body,
-                serde_json::json!({"name": "embed", "version": "fv_01K3EXACT"})
-            );
+            assert_eq!(body, serde_json::json!({"version": "fv_01K3EXACT"}));
            http::Response::builder().status(200).body(VERSION).unwrap()
        });
        let version = conn.get_function("embed", "fv_01K3EXACT").await.unwrap();
@@ -2843,21 +2853,20 @@ mod tests {
        let version: serde_json::Value = serde_json::from_str(VERSION).unwrap();
        let page = Arc::new(AtomicUsize::new(0));
        let conn = Connection::new_with_handler(move |request| {
-            assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/list");
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(body["include_definition"], true);
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/namespace/$/function/list");
+            let query = request.url().query_pairs().collect::<HashMap<_, _>>();
+            assert_eq!(query.get("include_definition").unwrap(), "true");
            match page.fetch_add(1, Ordering::SeqCst) {
                0 => {
-                    assert!(body.get("page_token").is_none());
+                    assert!(!query.contains_key("page_token"));
                    http::Response::builder()
                        .status(200)
                        .body(r#"{"functions": [], "page_token": "next"}"#.to_string())
                        .unwrap()
                }
                _ => {
-                    assert_eq!(body["page_token"], "next");
+                    assert_eq!(query.get("page_token").unwrap(), "next");
                    http::Response::builder()
                        .status(200)
                        .body(
@@ -2886,9 +2895,11 @@ mod tests {
        let seen = requests.clone();
        let conn = Connection::new_with_handler(move |request| {
            seen.fetch_add(1, Ordering::SeqCst);
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert!(body.get("page_token").is_none());
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/namespace/$/function/list");
+            let query = request.url().query_pairs().collect::<HashMap<_, _>>();
+            assert_eq!(query.get("include_definition").unwrap(), "true");
+            assert!(!query.contains_key("page_token"));
            http::Response::builder()
                .status(200)
                .body(r#"{"functions": [], "page_token": ""}"#)
@@ -2905,19 +2916,21 @@ mod tests {
        let page = Arc::new(AtomicUsize::new(0));
        let requests = page.clone();
        let conn = Connection::new_with_handler(move |request| {
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/namespace/$/function/list");
+            let query = request.url().query_pairs().collect::<HashMap<_, _>>();
+            assert_eq!(query.get("include_definition").unwrap(), "true");
            let next_page_token = match page.fetch_add(1, Ordering::SeqCst) {
                0 => {
-                    assert!(body.get("page_token").is_none());
+                    assert!(!query.contains_key("page_token"));
                    "one"
                }
                1 => {
-                    assert_eq!(body["page_token"], "one");
+                    assert_eq!(query.get("page_token").unwrap(), "one");
                    "two"
                }
                2 => {
-                    assert_eq!(body["page_token"], "two");
+                    assert_eq!(query.get("page_token").unwrap(), "two");
                    "one"
                }
                page => panic!("unexpected page: {page}"),
@@ -2952,13 +2965,10 @@ mod tests {
    async fn test_drop_function_sends_exact_version_and_decodes_replay() {
        let conn = Connection::new_with_handler(|request| {
            assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/drop");
+            assert_eq!(request.url().path(), "/v1/function/embed/drop");
            let body: serde_json::Value =
                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(
-                body,
-                serde_json::json!({"name": "embed", "version": "fv_01K3EXACT"})
-            );
+            assert_eq!(body, serde_json::json!({"version": "fv_01K3EXACT"}));
            http::Response::builder()
                .status(200)
                .body(r#"{"dropped":false}"#)