Implementation:Apache Paimon ApiResponse

Knowledge Sources	Apache_Paimon
Domains	REST API, Data Serialization
Last Updated	2026-02-08 00:00 GMT

Overview

ApiResponse provides response data structures for Apache Paimon's REST API interactions including error handling, paged results, and catalog metadata responses.

Description

The ApiResponse module defines a comprehensive set of dataclasses that represent different response types from the Paimon REST API. It includes base classes like RESTResponse and AuditRESTResponse, specialized responses for listing and retrieving databases and tables, error handling through ErrorResponse, and support for paginated results through PagedResponse and PagedList.

These response classes leverage the json_field decorator from pypaimon.common.json_util to map Python attributes to JSON field names, enabling seamless serialization and deserialization of REST API responses. The AuditRESTResponse base class provides audit trail fields (owner, created_at, created_by, updated_at, updated_by) that are inherited by database and table responses.

The module supports various catalog operations including GetTableResponse for table metadata retrieval, GetDatabaseResponse for database information, ConfigResponse for catalog configuration merging, and specialized responses for token management and commit operations. The PagedResponse implementations (ListDatabasesResponse, ListTablesResponse) enable efficient handling of large result sets through pagination tokens.

Usage

Use ApiResponse classes when implementing REST API clients for Paimon catalogs, deserializing API responses into typed Python objects, or building services that need to interact with Paimon's REST interface for metadata operations.

Code Reference

Source Location

• Repository: Apache_Paimon
• File: paimon-python/pypaimon/api/api_response.py

Signature

@dataclass
class PagedList(Generic[T]):
    elements: List[T]
    next_page_token: Optional[str] = None

class RESTResponse(ABC):
    """RESTResponse"""

@dataclass
class ErrorResponse(RESTResponse):
    resource_type: Optional[str] = json_field("resourceType", default=None)
    resource_name: Optional[str] = json_field("resourceName", default=None)
    message: Optional[str] = json_field("message", default=None)
    code: Optional[int] = json_field("code", default=None)

@dataclass
class AuditRESTResponse(RESTResponse):
    owner: Optional[str] = json_field("owner", default=None)
    created_at: Optional[int] = json_field("createdAt", default=None)
    created_by: Optional[str] = json_field("createdBy", default=None)
    updated_at: Optional[int] = json_field("updatedAt", default=None)
    updated_by: Optional[str] = json_field("updatedBy", default=None)

class PagedResponse(RESTResponse, Generic[T]):
    @abstractmethod
    def data(self) -> List[T]:
        pass

    @abstractmethod
    def get_next_page_token(self) -> str:
        pass

@dataclass
class GetTableResponse(AuditRESTResponse):
    id: Optional[str]
    name: Optional[str]
    path: Optional[str]
    is_external: Optional[bool]
    schema_id: Optional[int]
    schema: Optional[Schema]

@dataclass
class GetDatabaseResponse(AuditRESTResponse):
    id: Optional[str]
    name: Optional[str]
    location: Optional[str]
    options: Optional[Dict[str, str]]

Import

from pypaimon.api.api_response import (
    RESTResponse, ErrorResponse, AuditRESTResponse,
    PagedResponse, PagedList, GetTableResponse,
    GetDatabaseResponse, ListDatabasesResponse,
    ListTablesResponse, ConfigResponse
)

I/O Contract

Inputs

Name	Type	Required	Description
resource_type	str	No	Type of the resource in error response
resource_name	str	No	Name of the resource in error response
message	str	No	Error message description
code	int	No	HTTP error code
owner	str	No	Owner of the resource (audit field)
created_at	int	No	Creation timestamp in milliseconds (audit field)

Outputs

Name	Type	Description
ErrorResponse	ErrorResponse	Error details from REST API
GetTableResponse	GetTableResponse	Table metadata including schema
GetDatabaseResponse	GetDatabaseResponse	Database metadata and options
ListDatabasesResponse	ListDatabasesResponse	Paginated list of database names
ListTablesResponse	ListTablesResponse	Paginated list of table names

Usage Examples

from pypaimon.api.api_response import GetTableResponse, ErrorResponse
from pypaimon.common.json_util import JSON

# Deserialize a table response from JSON
json_response = '{"id": "table-123", "name": "my_table", ...}'
table_response = JSON.from_json(json_response, GetTableResponse)

# Access table metadata
table_id = table_response.get_id()
table_name = table_response.get_name()
schema = table_response.get_schema()

# Handle error responses
error = ErrorResponse(
    resource_type="Table",
    resource_name="my_table",
    message="Table not found",
    code=404
)

# Work with paged results
from pypaimon.api.api_response import ListDatabasesResponse
db_list = ListDatabasesResponse(
    databases=["db1", "db2", "db3"],
    next_page_token="next_page_token_value"
)
for db in db_list.data():
    print(f"Database: {db}")

Related Pages

• Principle:Apache_Paimon_REST_Catalog_Protocol