[python] Support REST function API in pypaimon (#7559)

Author: XiaoHongbo-Hope

paimon-python/pypaimon/api/api_request.py

@@ -22,6 +22,9 @@
 
 from pypaimon.common.identifier import Identifier
 from pypaimon.common.json_util import json_field
+from pypaimon.function.function_change import FunctionChange
+from pypaimon.function.function_definition import FunctionDefinition
+from pypaimon.schema.data_types import DataField
 from pypaimon.schema.schema import Schema
 from pypaimon.schema.schema_change import SchemaChange
 from pypaimon.snapshot.snapshot import Snapshot
@@ -94,3 +97,62 @@ class RollbackTableRequest(RESTRequest):
 
     instant: Instant = json_field(FIELD_INSTANT)
     from_snapshot: Optional[int] = json_field(FIELD_FROM_SNAPSHOT)
+
+
+@dataclass
+class CreateFunctionRequest(RESTRequest):
+    FIELD_NAME = "name"
+    FIELD_INPUT_PARAMS = "inputParams"
+    FIELD_RETURN_PARAMS = "returnParams"
+    FIELD_DETERMINISTIC = "deterministic"
+    FIELD_DEFINITIONS = "definitions"
+    FIELD_COMMENT = "comment"
+    FIELD_OPTIONS = "options"
+
+    name: str = json_field(FIELD_NAME)
+    input_params: Optional[List[DataField]] = json_field(FIELD_INPUT_PARAMS, default=None)
+    return_params: Optional[List[DataField]] = json_field(FIELD_RETURN_PARAMS, default=None)
+    deterministic: bool = json_field(FIELD_DETERMINISTIC, default=False)
+    definitions: Optional[Dict[str, FunctionDefinition]] = json_field(FIELD_DEFINITIONS, default=None)
+    comment: Optional[str] = json_field(FIELD_COMMENT, default=None)
+    options: Optional[Dict[str, str]] = json_field(FIELD_OPTIONS, default=None)
+
+    def to_dict(self) -> Dict:
+        result = {
+            self.FIELD_NAME: self.name,
+            self.FIELD_DETERMINISTIC: self.deterministic,
+        }
+        if self.input_params is not None:
+            result[self.FIELD_INPUT_PARAMS] = [
+                p.to_dict() if hasattr(p, 'to_dict') else p for p in self.input_params
+            ]
+        else:
+            result[self.FIELD_INPUT_PARAMS] = None
+        if self.return_params is not None:
+            result[self.FIELD_RETURN_PARAMS] = [
+                p.to_dict() if hasattr(p, 'to_dict') else p for p in self.return_params
+            ]
+        else:
+            result[self.FIELD_RETURN_PARAMS] = None
+        if self.definitions is not None:
+            result[self.FIELD_DEFINITIONS] = {
+                k: v.to_dict() if hasattr(v, 'to_dict') else v
+                for k, v in self.definitions.items()
+            }
+        else:
+            result[self.FIELD_DEFINITIONS] = None
+        result[self.FIELD_COMMENT] = self.comment
+        result[self.FIELD_OPTIONS] = self.options
+        return result
+
+
+@dataclass
+class AlterFunctionRequest(RESTRequest):
+    FIELD_CHANGES = "changes"
+
+    changes: List[FunctionChange] = json_field(FIELD_CHANGES)
+
+    def to_dict(self) -> Dict:
+        return {
+            self.FIELD_CHANGES: [c.to_dict() for c in self.changes]
+        }

paimon-python/pypaimon/api/api_response.py

@@ -20,8 +20,10 @@
 from dataclasses import dataclass
 from typing import Dict, Generic, List, Optional
 
+from pypaimon.common.identifier import Identifier
 from pypaimon.common.json_util import T, json_field
 from pypaimon.common.options import Options
+from pypaimon.schema.data_types import DataField
 from pypaimon.schema.schema import Schema
 from pypaimon.snapshot.snapshot_commit import PartitionStatistics
 from pypaimon.snapshot.table_snapshot import TableSnapshot
@@ -327,3 +329,227 @@ def __init__(self, snapshot: Optional[TableSnapshot] = None):
 
     def get_snapshot(self) -> Optional[TableSnapshot]:
         return self.snapshot
+
+
+@dataclass
+class GetFunctionResponse(AuditRESTResponse):
+    """Response for getting a function."""
+    FIELD_UUID = "uuid"
+    FIELD_NAME = "name"
+    FIELD_INPUT_PARAMS = "inputParams"
+    FIELD_RETURN_PARAMS = "returnParams"
+    FIELD_DETERMINISTIC = "deterministic"
+    FIELD_DEFINITIONS = "definitions"
+    FIELD_COMMENT = "comment"
+    FIELD_OPTIONS = "options"
+
+    uuid: Optional[str] = json_field(FIELD_UUID, default=None)
+    name: Optional[str] = json_field(FIELD_NAME, default=None)
+    input_params: Optional[List[DataField]] = json_field(FIELD_INPUT_PARAMS, default=None)
+    return_params: Optional[List[DataField]] = json_field(FIELD_RETURN_PARAMS, default=None)
+    deterministic: bool = json_field(FIELD_DETERMINISTIC, default=False)
+    definitions: Optional[Dict[str, 'FunctionDefinition']] = json_field(FIELD_DEFINITIONS, default=None)
+    comment: Optional[str] = json_field(FIELD_COMMENT, default=None)
+    options: Optional[Dict[str, str]] = json_field(FIELD_OPTIONS, default=None)
+
+    def __init__(
+        self,
+        uuid: Optional[str] = None,
+        name: Optional[str] = None,
+        input_params: Optional[List[DataField]] = None,
+        return_params: Optional[List[DataField]] = None,
+        deterministic: bool = False,
+        definitions: Optional[Dict[str, 'FunctionDefinition']] = None,
+        comment: Optional[str] = None,
+        options: Optional[Dict[str, str]] = None,
+        owner: Optional[str] = None,
+        created_at: Optional[int] = None,
+        created_by: Optional[str] = None,
+        updated_at: Optional[int] = None,
+        updated_by: Optional[str] = None,
+    ):
+        super().__init__(owner, created_at, created_by, updated_at, updated_by)
+        self.uuid = uuid
+        self.name = name
+        self.input_params = input_params
+        self.return_params = return_params
+        self.deterministic = deterministic
+        self.definitions = definitions
+        self.comment = comment
+        self.options = options
+
+    def to_function(self, identifier):
+        from pypaimon.function.function import FunctionImpl
+        return FunctionImpl(
+            identifier=identifier,
+            input_params=self.input_params,
+            return_params=self.return_params,
+            deterministic=self.deterministic,
+            definitions=self.definitions or {},
+            comment=self.comment,
+            options=self.options or {},
+        )
+
+    @staticmethod
+    def _parse_data_fields(raw: Optional[list]) -> Optional[List[DataField]]:
+        if raw is None:
+            return None
+        return [DataField.from_dict(f) if isinstance(f, dict) else f for f in raw]
+
+    @staticmethod
+    def _parse_definitions(raw) -> Optional[Dict[str, 'FunctionDefinition']]:
+        from pypaimon.function.function_definition import FunctionDefinition
+        if raw is None:
+            return None
+        return {
+            k: FunctionDefinition.from_dict(v) if isinstance(v, dict) else v
+            for k, v in raw.items()
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> "GetFunctionResponse":
+        return cls(
+            uuid=data.get("uuid"),
+            name=data.get("name"),
+            input_params=cls._parse_data_fields(data.get("inputParams")),
+            return_params=cls._parse_data_fields(data.get("returnParams")),
+            deterministic=data.get("deterministic", False),
+            definitions=cls._parse_definitions(data.get("definitions")),
+            comment=data.get("comment"),
+            options=data.get("options"),
+            owner=data.get("owner"),
+            created_at=data.get("createdAt"),
+            created_by=data.get("createdBy"),
+            updated_at=data.get("updatedAt"),
+            updated_by=data.get("updatedBy"),
+        )
+
+    def to_dict(self) -> Dict:
+        result = {}
+        if self.uuid is not None:
+            result["uuid"] = self.uuid
+        result["name"] = self.name
+        result["inputParams"] = (
+            [p.to_dict() if hasattr(p, 'to_dict') else p for p in self.input_params]
+            if self.input_params is not None else None
+        )
+        result["returnParams"] = (
+            [p.to_dict() if hasattr(p, 'to_dict') else p for p in self.return_params]
+            if self.return_params is not None else None
+        )
+        result["deterministic"] = self.deterministic
+        if self.definitions is not None:
+            result["definitions"] = {
+                k: v.to_dict() if hasattr(v, 'to_dict') else v
+                for k, v in self.definitions.items()
+            }
+        else:
+            result["definitions"] = None
+        result["comment"] = self.comment
+        result["options"] = self.options
+        if self.owner is not None:
+            result["owner"] = self.owner
+        if self.created_at is not None:
+            result["createdAt"] = self.created_at
+        if self.created_by is not None:
+            result["createdBy"] = self.created_by
+        if self.updated_at is not None:
+            result["updatedAt"] = self.updated_at
+        if self.updated_by is not None:
+            result["updatedBy"] = self.updated_by
+        return result
+
+
+@dataclass
+class ListFunctionsResponse(PagedResponse[str]):
+    """Response for listing functions."""
+    FIELD_FUNCTIONS = "functions"
+
+    functions: Optional[List[str]] = json_field(FIELD_FUNCTIONS, default=None)
+    next_page_token: Optional[str] = json_field(
+        PagedResponse.FIELD_NEXT_PAGE_TOKEN, default=None)
+
+    def data(self) -> Optional[List[str]]:
+        return self.functions
+
+    def get_next_page_token(self) -> Optional[str]:
+        return self.next_page_token
+
+
+@dataclass
+class ListFunctionDetailsResponse(PagedResponse['GetFunctionResponse']):
+    """Response for listing function details."""
+    FIELD_FUNCTION_DETAILS = "functionDetails"
+
+    function_details: Optional[List[GetFunctionResponse]] = json_field(
+        FIELD_FUNCTION_DETAILS, default=None)
+    next_page_token: Optional[str] = json_field(
+        PagedResponse.FIELD_NEXT_PAGE_TOKEN, default=None)
+
+    def data(self) -> Optional[List[GetFunctionResponse]]:
+        return self.function_details
+
+    def get_next_page_token(self) -> Optional[str]:
+        return self.next_page_token
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> "ListFunctionDetailsResponse":
+        details = data.get("functionDetails")
+        if details is not None:
+            details = [GetFunctionResponse.from_dict(d) for d in details]
+        return cls(
+            function_details=details,
+            next_page_token=data.get("nextPageToken"),
+        )
+
+    def to_dict(self) -> Dict:
+        result = {}
+        if self.function_details is not None:
+            result["functionDetails"] = [d.to_dict() for d in self.function_details]
+        else:
+            result["functionDetails"] = None
+        result["nextPageToken"] = self.next_page_token
+        return result
+
+
+@dataclass
+class ListFunctionsGloballyResponse(PagedResponse[Identifier]):
+    """Response for listing functions globally across databases."""
+    FIELD_FUNCTIONS = "functions"
+
+    functions: Optional[List[Identifier]] = json_field(FIELD_FUNCTIONS, default=None)
+    next_page_token: Optional[str] = json_field(
+        PagedResponse.FIELD_NEXT_PAGE_TOKEN, default=None)
+
+    def data(self) -> Optional[List[Identifier]]:
+        return self.functions
+
+    def get_next_page_token(self) -> Optional[str]:
+        return self.next_page_token
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> "ListFunctionsGloballyResponse":
+        functions = data.get("functions")
+        if functions is not None:
+            functions = [
+                Identifier.from_string(f) if isinstance(f, str) else
+                Identifier.create(f.get("database"), f.get("object"))
+                if isinstance(f, dict) else f
+                for f in functions
+            ]
+        return cls(
+            functions=functions,
+            next_page_token=data.get("nextPageToken"),
+        )
+
+    def to_dict(self) -> Dict:
+        result = {}
+        if self.functions is not None:
+            result["functions"] = [
+                {"database": f.get_database_name(), "object": f.get_object_name()}
+                for f in self.functions
+            ]
+        else:
+            result["functions"] = None
+        result["nextPageToken"] = self.next_page_token
+        return result

paimon-python/pypaimon/api/resource_paths.py

@@ -28,6 +28,8 @@ class ResourcePaths:
     TABLES = "tables"
     TABLE_DETAILS = "table-details"
     PARTITIONS = "partitions"
+    FUNCTIONS = "functions"
+    FUNCTION_DETAILS = "function-details"
 
     def __init__(self, prefix: str):
         self.base_path = "/{}/{}".format(self.V1, prefix).rstrip("/")
@@ -83,3 +85,17 @@ def table_snapshot(self, database_name: str, table_name: str) -> str:
     def partitions(self, database_name: str, table_name: str) -> str:
         return ("{}/{}/{}/{}/{}/{}".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
                                            self.TABLES, RESTUtil.encode_string(table_name), self.PARTITIONS))
+
+    def functions(self, database_name: Optional[str] = None) -> str:
+        if database_name:
+            return "{}/{}/{}/{}".format(self.base_path, self.DATABASES,
+                                        RESTUtil.encode_string(database_name), self.FUNCTIONS)
+        return "{}/{}".format(self.base_path, self.FUNCTIONS)
+
+    def function_details(self, database_name: str) -> str:
+        return "{}/{}/{}/{}".format(self.base_path, self.DATABASES,
+                                    RESTUtil.encode_string(database_name), self.FUNCTION_DETAILS)
+
+    def function(self, database_name: str, function_name: str) -> str:
+        return "{}/{}/{}/{}/{}".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
+                                       self.FUNCTIONS, RESTUtil.encode_string(function_name))