This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6ff82e8295 [python] Align RESTApi and openapi-v4 actions with the
OpenAPI spec, checked in CI (#10623)
6ff82e8295 is described below
commit 6ff82e8295f4dd170e9b6538dcfc59420bb5d2e9
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sun Oct 18 09:23:27 2026 +0800
[python] Align RESTApi and openapi-v4 actions with the OpenAPI spec,
checked in CI (#10623)
---
.../pypaimon/api/auth/dlf_openapi_actions.py | 30 +++-
.../pypaimon/tests/rest/dlf_signer_test.py | 27 ++-
.../pypaimon/tests/rest/rest_api_coverage_test.py | 199 +++++++++++++++++++++
tools/ci/plan.py | 9 +
tools/ci/test_plan.py | 11 ++
5 files changed, 268 insertions(+), 8 deletions(-)
diff --git a/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py
b/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py
index 8fd46e9f63..e6e1163600 100644
--- a/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py
+++ b/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py
@@ -31,21 +31,32 @@ OPERATIONS: Tuple[Tuple[str, str, str], ...] = (
("GET", "/v1/{prefix}/databases/{database}/tables", "ListTables"),
("POST", "/v1/{prefix}/databases/{database}/tables", "CreateTable"),
("GET", "/v1/{prefix}/databases/{database}/table-details",
"ListTableDetails"),
+ ("GET", "/v1/{prefix}/tables", "ListTablesGlobally"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}", "GetTable"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}", "AlterTable"),
("DELETE", "/v1/{prefix}/databases/{database}/tables/{table}",
"DropTable"),
+ ("GET", "/v1/{prefix}/tables/id/{tableId}", "GetTableById"),
("POST", "/v1/{prefix}/tables/rename", "RenameTable"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/auth",
"AuthTableQuery"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit",
"CommitTable"),
+ ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit-lock",
"AcquireCommitLock"),
+ ("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/commit-lock/renew",
+ "RenewCommitLock"),
+ ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/replace",
"ReplaceTable"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/token",
"GetTableToken"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/rollback",
"RollbackToSnapshot"),
+ ("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/rollback-schema",
"RollbackSchema"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/snapshot",
"GetTableSnapshot"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/snapshots",
"ListSnapshots"),
- ("GET",
"/v1/{prefix}/databases/{database}/tables/{table}/snapshots/{version}",
"GetVersionSnapshot"),
+ ("GET",
"/v1/{prefix}/databases/{database}/tables/{table}/snapshots/{version}",
+ "GetVersionSnapshot"),
+ ("GET", "/v1/{prefix}/databases/{database}/tables/{table}/schemas",
"ListSchemas"),
+ ("GET",
"/v1/{prefix}/databases/{database}/tables/{table}/schemas/{version}",
"GetSchema"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/partitions",
"ListPartitions"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/partitions",
"CreatePartitions"),
("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/partitions/drop",
"DropPartitions"),
- ("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/partitions/mark",
"MarkDonePartitions"),
+ ("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/partitions/mark",
+ "MarkDonePartitions"),
("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/partitions/list-by-names",
"ListPartitionsByNames"),
("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/partitions/list-by-filter",
@@ -59,24 +70,39 @@ OPERATIONS: Tuple[Tuple[str, str, str], ...] = (
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/tags",
"CreateTag"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}",
"GetTag"),
("DELETE", "/v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}",
"DropTag"),
+ ("GET", "/v1/{prefix}/databases/{database}/tables/{table}/consumers",
"ListConsumers"),
+ ("POST",
"/v1/{prefix}/databases/{database}/tables/{table}/consumers/reset",
"ResetConsumer"),
("GET", "/v1/{prefix}/databases/{database}/tables/{table}/policies",
"ListPolicies"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/policies",
"CreatePolicy"),
("POST", "/v1/{prefix}/databases/{database}/tables/{table}/policies/drop",
"DropPolicy"),
("GET", "/v1/{prefix}/databases/{database}/views", "ListViews"),
("POST", "/v1/{prefix}/databases/{database}/views", "CreateView"),
("GET", "/v1/{prefix}/databases/{database}/view-details",
"ListViewDetails"),
+ ("GET", "/v1/{prefix}/views", "ListViewsGlobally"),
("GET", "/v1/{prefix}/databases/{database}/views/{view}", "GetView"),
("POST", "/v1/{prefix}/databases/{database}/views/{view}", "AlterView"),
("DELETE", "/v1/{prefix}/databases/{database}/views/{view}", "DropView"),
("POST", "/v1/{prefix}/views/rename", "RenameView"),
+ ("GET", "/v1/{prefix}/databases/{database}/semantic-views",
"ListSemanticViews"),
+ ("GET", "/v1/{prefix}/databases/{database}/semantic-views/{semanticView}",
"GetSemanticView"),
+ ("POST", "/v1/{prefix}/databases/{database}/semantic-views/{semanticView}",
+ "UpsertSemanticView"),
+ ("DELETE",
"/v1/{prefix}/databases/{database}/semantic-views/{semanticView}",
+ "DeleteSemanticView"),
("GET", "/v1/{prefix}/databases/{database}/functions", "ListFunctions"),
("POST", "/v1/{prefix}/databases/{database}/functions", "CreateFunction"),
+ ("GET", "/v1/{prefix}/databases/{database}/function-details",
"ListFunctionDetails"),
+ ("GET", "/v1/{prefix}/functions", "ListFunctionsGlobally"),
("GET", "/v1/{prefix}/databases/{database}/functions/{function}",
"GetFunction"),
("POST", "/v1/{prefix}/databases/{database}/functions/{function}",
"AlterFunction"),
("DELETE", "/v1/{prefix}/databases/{database}/functions/{function}",
"DropFunction"),
("GET", "/v1/{prefix}/permissions", "ListPermissionAssignments"),
("POST", "/v1/{prefix}/permissions/grant", "GrantPermissionAssignment"),
("POST", "/v1/{prefix}/permissions/revoke", "RevokePermissionAssignment"),
+ ("GET", "/v1/{prefix}/labels", "ListLabels"),
+ ("POST", "/v1/{prefix}/labels/get", "GetLabel"),
+ ("POST", "/v1/{prefix}/labels/upsert", "UpsertLabel"),
+ ("POST", "/v1/{prefix}/labels/delete", "DeleteLabel"),
)
diff --git a/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
b/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
index 17e380c7ca..65e766c292 100644
--- a/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
+++ b/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
@@ -18,6 +18,7 @@
import unittest
import hashlib
import locale
+import os
import re
import threading
from datetime import datetime, timedelta, timezone
@@ -476,9 +477,9 @@ class DLFSignerTest(unittest.TestCase):
self.assertEqual(
"ACS4-HMAC-SHA256
Credential=TestAKId/20250416/cn-hangzhou/DlfNext/aliyun_v4_request,"
-
"SignedHeaders=host;x-acs-content-sha256;x-acs-date;x-acs-signature-nonce;"
- "x-acs-version,"
-
"Signature=9812bacd92dd1152e7ea32aacfad0545835d6a18543852364f1a1ea0cba6219a",
+ "SignedHeaders=host;x-acs-action;x-acs-content-sha256;x-acs-date;"
+ "x-acs-signature-nonce;x-acs-version,"
+
"Signature=2d5fdb5201901a5cdceee97318366666a88504eeae0b1ce534dbf86889cc00fb",
signer.authorization(rest_param, token, host, headers))
def test_openapi_v4_percent_encode(self):
@@ -501,7 +502,19 @@ class DLFSignerTest(unittest.TestCase):
path = re.sub(r"\{[^}]+\}", "x1", template)
self.assertEqual(action, resolve_action(method, path), path)
actions.add(action)
- self.assertEqual(53, len(actions))
+ self.assertEqual(74, len(actions))
+
+ def test_openapi_v4_operations_match_java(self):
+ """pypaimon registers Java's operations, in Java's order, under the
same actions."""
+ source = os.path.join(os.path.dirname(os.path.abspath(__file__)),
*[os.pardir] * 4, "paimon-api",
+
"src/main/java/org/apache/paimon/rest/auth/DLFOpenApiActions.java")
+ with open(source, encoding="utf-8") as f:
+ java = re.findall(r'\{\s*"(\w+)",\s*"([^"]+)",\s*"(\w+)"\s*\}',
f.read())
+ self.assertEqual(java, list(OPERATIONS))
+
+ def test_openapi_v4_action_names_are_unique(self):
+ names = [action for _, _, action in OPERATIONS]
+ self.assertEqual([], sorted({name for name in names if
names.count(name) > 1}))
def test_openapi_v4_client_paths_resolve_to_registered_actions(self):
paths = ResourcePaths("clg-paimon-1")
@@ -524,6 +537,8 @@ class DLFSignerTest(unittest.TestCase):
("GetTableSnapshot", "GET", paths.table_snapshot("db", "t")),
("ListPartitions", "GET", paths.partitions("db", "t")),
("ListFunctions", "GET", paths.functions("db")),
+ ("ListFunctionDetails", "GET", paths.function_details("db")),
+ ("ListFunctionsGlobally", "GET", paths.functions()),
("GetFunction", "GET", paths.function("db", "f")),
("ListTags", "GET", paths.tags("db", "t")),
("GetTag", "GET", paths.tag("db", "t", "tag")),
@@ -547,7 +562,7 @@ class DLFSignerTest(unittest.TestCase):
def test_openapi_v4_unregistered_requests_have_no_action(self):
paths = ResourcePaths("clg-paimon-1")
- self.assertIsNone(resolve_action("GET", paths.tables()))
+ self.assertIsNone(resolve_action("GET", paths.rename_table()))
self.assertIsNone(resolve_action("POST", paths.branch("db", "t", "b")
+ "/rename"))
self.assertIsNone(resolve_action("PUT", paths.table("db", "t")))
self.assertIsNone(resolve_action("GET", ResourcePaths.config() + "/"))
@@ -582,7 +597,7 @@ class DLFSignerTest(unittest.TestCase):
token = DLFToken("TestAKId", "TestAKSecret", None, None)
host = "dlfnext.cn-hangzhou.aliyuncs.com"
now = datetime(2025, 4, 16, 3, 44, 46, tzinfo=timezone.utc)
- rest_param = RESTAuthParameter("GET", "/v1/clg-paimon-1/tables", "",
{})
+ rest_param = RESTAuthParameter("GET",
"/v1/clg-paimon-1/databases/db/tables/t/unknown", "", {})
headers = signer.sign_headers(None, now, None, host)
authorization = signer.authorization(rest_param, token, host, headers)
diff --git a/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py
b/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py
new file mode 100644
index 0000000000..b55e0da0dd
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py
@@ -0,0 +1,199 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import inspect
+import os
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import Dict, List
+from urllib.parse import urlsplit
+
+import yaml
+
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.common.identifier import Identifier
+from pypaimon.function.function import FunctionImpl
+from pypaimon.function.function_change import FunctionChange
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.list_permissions_request import ListPermissionsRequest
+from pypaimon.management.list_policies_request import ListPoliciesRequest
+from pypaimon.management.permission_assignment import PermissionAssignment
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+from pypaimon.management.resource_type import ResourceType
+from pypaimon.management.row_filter import RowFilter
+from pypaimon.schema.schema import Schema
+from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.snapshot.snapshot_commit import PartitionStatistics
+from pypaimon.table.instant import Instant
+
+# RESTApi methods with no entry in dlf_openapi_actions; they are signed
without an action.
+WITHOUT_ACTION = frozenset()
+
+_SPEC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)),
*[os.pardir] * 4, "docs", "static")
+_SPEC_FILES = ("rest-catalog-open-api.yaml", "rest-management-open-api.yaml")
+_HTTP_METHODS = ("get", "put", "post", "delete", "patch", "head")
+
+_TABLE = PermissionResource(ResourceType.TABLE, "db", "t", None, None)
+
+# A value per parameter annotation, else per name; "SELECT" is valid both as a
name and as an access.
+_ARGUMENTS = {
+ str: "SELECT",
+ bool: False,
+ Identifier: Identifier.create("db", "t"),
+ Schema: Schema(),
+ List: [FunctionChange.set_option("k", "v")], # serializes for alter_table
and alter_function
+ List[Dict[str, str]]: [{"dt": "1"}],
+ List[PartitionStatistics]: [],
+ Snapshot: Snapshot(version=3, id=1, schema_id=0, base_manifest_list="b",
delta_manifest_list="d",
+ total_record_count=1, delta_record_count=1,
commit_user="u",
+ commit_identifier=1, commit_kind="APPEND",
time_millis=0),
+ PermissionResource: _TABLE,
+ PermissionAssignment: PermissionAssignment(_TABLE, "SELECT", "x1"),
+ ListPermissionsRequest: ListPermissionsRequest(ResourceType.TABLE, "db",
"t"),
+ ListPoliciesRequest: ListPoliciesRequest(_TABLE),
+ PolicyType: PolicyType.ROW_FILTER,
+ DataPolicy: DataPolicy.row_filter(_TABLE, RowFilter("x1"), "x1"),
+ "identifier": Identifier.create("db", "t"),
+ "instant": Instant.snapshot(1),
+ "function": FunctionImpl(Identifier.create("db", "f")),
+}
+
+
+class RESTApiCoverageTest(unittest.TestCase):
+ """Every request RESTApi sends must be in the OpenAPI spec and carry an
openapi-v4 action."""
+
+ def setUp(self):
+ requests = self.requests = []
+
+ class Handler(BaseHTTPRequestHandler):
+
+ def log_message(self, *args):
+ pass
+
+ def record(self):
+ self.rfile.read(int(self.headers.get("Content-Length", 0)))
+ requests.append((self.command, urlsplit(self.path).path,
self.headers.get("x-acs-action")))
+ self.send_response(200)
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", "2")
+ self.end_headers()
+ self.wfile.write(b"{}")
+
+ do_GET = do_POST = do_DELETE = record
+
+ self.server = HTTPServer(("127.0.0.1", 0), Handler)
+ threading.Thread(target=self.server.serve_forever, daemon=True).start()
+ self.options = {
+ "uri": "http://127.0.0.1:{}".format(self.server.server_port),
+ "prefix": "clg-paimon-1",
+ "token.provider": "dlf",
+ "dlf.access-key-id": "ak",
+ "dlf.access-key-secret": "sk",
+ "dlf.region": "cn-hangzhou",
+ "dlf.signing-algorithm": "openapi-v4",
+ }
+
+ def tearDown(self):
+ self.server.shutdown()
+ self.server.server_close()
+
+ def test_every_request_carries_an_action(self):
+ """A new REST call fails here until it is added to dlf_openapi_actions
or WITHOUT_ACTION."""
+ without_action = {}
+ for name, requests in self._send_every_request().items():
+ for verb, path, action in requests:
+ if action is None:
+ without_action.setdefault(name, []).append(verb + " " +
path)
+
+ missing = {name: paths for name, paths in without_action.items() if
name not in WITHOUT_ACTION}
+ self.assertEqual({}, missing, "requests without x-acs-action; map them
in dlf_openapi_actions")
+ self.assertEqual(set(), WITHOUT_ACTION - set(without_action),
+ "WITHOUT_ACTION entries that now resolve; remove
them")
+
+ def test_every_request_is_in_the_openapi_spec(self):
+ """A new REST call fails here until docs/static/rest-*-open-api.yaml
lists it."""
+ spec = _spec_operations()
+ unspecified = {}
+ for name, requests in self._send_every_request().items():
+ for verb, path, _ in requests:
+ if not _in_spec(spec, verb, path):
+ unspecified.setdefault(name, []).append(verb + " " + path)
+
+ self.assertEqual({}, unspecified, "requests the OpenAPI spec lacks;
add them to it")
+
+ def _send_every_request(self):
+ """Requests sent by each public method, by name; "RESTApi" holds the
constructor's."""
+ try:
+ RESTApi(dict(self.options, warehouse="x1"))
+ except Exception:
+ pass # a reply of {} may not parse; only the request matters
+ sent = {"RESTApi": list(self.requests)}
+ silent = [] if self.requests else ["RESTApi"]
+ api = RESTApi(self.options, False)
+ for name, method in sorted(vars(RESTApi).items()):
+ if name.startswith("_") or name == "get_options" or not
inspect.isfunction(method):
+ continue
+ del self.requests[:]
+ try:
+ getattr(api, name)(*_arguments(method))
+ except Exception:
+ pass # a reply of {} may not parse; only the request matters
+ if not self.requests:
+ silent.append(name)
+ sent[name] = list(self.requests)
+
+ self.assertEqual([], silent, "methods that sent no request; give them
arguments")
+ return sent
+
+
+def _spec_operations():
+ operations = set()
+ for file in _SPEC_FILES:
+ with open(os.path.join(_SPEC_DIR, file), encoding="utf-8") as f:
+ for path, item in yaml.safe_load(f)["paths"].items():
+ operations.update((method.upper(), path) for method in item if
method in _HTTP_METHODS)
+ return operations
+
+
+def _in_spec(spec, verb, path):
+ segments = path.split("/")
+ for method, template in spec:
+ parts = template.split("/")
+ if method == verb and len(parts) == len(segments) and all(
+ segment != "" if part.startswith("{") else part == segment
+ for part, segment in zip(parts, segments)):
+ return True
+ return False
+
+
+def _arguments(method):
+ arguments = []
+ for parameter in list(inspect.signature(method).parameters.values())[1:]:
+ if parameter.default is not inspect.Parameter.empty:
+ continue
+ annotation = parameter.annotation
+ if type(None) in getattr(annotation, "__args__", ()):
+ arguments.append(None)
+ else:
+ arguments.append(_ARGUMENTS.get(annotation,
_ARGUMENTS.get(parameter.name)))
+ return arguments
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tools/ci/plan.py b/tools/ci/plan.py
index b539f45a60..49954ca2bb 100644
--- a/tools/ci/plan.py
+++ b/tools/ci/plan.py
@@ -41,6 +41,9 @@ SUITES = [
("full-text", "Full text / JDK 8", 8, "2.12", 30),
]
ALL_JAVA = {suite[0] for suite in SUITES}
+# Java's core tests and pypaimon's tests both check the REST client against
these.
+REST_SPECS = {"docs/static/rest-catalog-open-api.yaml",
"docs/static/rest-management-open-api.yaml"}
+OPENAPI_V4_ACTIONS =
"paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFOpenApiActions.java"
FLINK = {"flink1-common", "flink1-connectors", "flink2"}
SPARK = {"spark3", "spark4"}
E2E = {"e2e-flink1", "e2e-flink2"}
@@ -54,6 +57,10 @@ def select(paths, full=False):
docs = python = licensing = True
for path in paths:
+ if path in REST_SPECS:
+ docs = python = True
+ selected.add("core")
+ continue
if path.startswith("docs/"):
docs = True
continue
@@ -80,6 +87,8 @@ def select(paths, full=False):
continue
licensing = True
+ if path == OPENAPI_V4_ACTIONS:
+ python = True
if path.startswith("paimon-spark/"):
# Core includes spark-common tests and paimon-docs completeness
tests.
# SparkE2eTest is disabled on JDK 11, so only the Flink 1 E2E lane
applies.
diff --git a/tools/ci/test_plan.py b/tools/ci/test_plan.py
index 4d0c26a174..afcdb51e0c 100644
--- a/tools/ci/test_plan.py
+++ b/tools/ci/test_plan.py
@@ -79,6 +79,17 @@ class SelectionTest(unittest.TestCase):
self.assertTrue(result['python'])
self.assertFalse(result['java'])
+ def test_rest_contract_runs_java_and_python(self):
+ for path in sorted(plan.REST_SPECS):
+ with self.subTest(path=path):
+ result = plan.select([path])
+ self.assertTrue(result['docs'] and result['python'])
+ self.assertEqual(self.suites([path]), {'core'})
+ result = plan.select([plan.OPENAPI_V4_ACTIONS])
+ self.assertTrue(result['python'] and result['licensing'])
+ self.assertEqual(self.suites([plan.OPENAPI_V4_ACTIONS]), plan.ALL_JAVA)
+
self.assertFalse(plan.select(['paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java'])['python'])
+
def test_leaf_modules_and_workflows(self):
self.assertEqual(self.suites(['paimon-eslib/pom.xml']), {'eslib'})
self.assertEqual(self.suites(['paimon-e2e-tests/pom.xml']),
