Implementation:Apache Paimon TokenLoader

Knowledge Sources	Apache_Paimon
Domains	Authentication, Cloud Integration
Last Updated	2026-02-08 00:00 GMT

Overview

TokenLoader provides DLF token management including token representation, ECS metadata service integration for automatic token loading, and factory creation with HTTP retry support.

Description

The TokenLoader module contains classes for managing Alibaba Cloud DLF (Data Lake Formation) authentication tokens. The DLFToken dataclass represents a token with access key ID, secret, security token, and expiration information. It supports both string-based expiration timestamps and millisecond precision expiration times, automatically parsing ISO 8601 datetime strings into millisecond epochs.

DLFTokenLoader defines the abstract interface for loading tokens, with DLFECSTokenLoader providing a concrete implementation that fetches tokens from ECS (Elastic Compute Service) metadata service. The ECS loader automatically retrieves IAM role credentials from the metadata endpoint at http://100.100.100.200/latest/meta-data/Ram/security-credentials/, making it suitable for applications running on Alibaba Cloud ECS instances.

The module includes HTTPClient, a wrapper around requests.Session with exponential retry and configurable timeouts for reliable HTTP operations. DLFTokenLoaderFactory provides a simple factory method to create token loaders from catalog options. The ECS token loader supports both automatic role detection and explicit role name configuration, with comprehensive error handling for network failures and malformed responses.

Usage

Use TokenLoader when implementing DLF authentication for Paimon catalogs on Alibaba Cloud, loading temporary credentials from ECS metadata service, or building token refresh mechanisms for long-running applications.

Code Reference

Source Location

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

Signature

@dataclass
class DLFToken:
    TOKEN_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"

    access_key_id: str = json_field('AccessKeyId')
    access_key_secret: str = json_field('AccessKeySecret')
    security_token: Optional[str] = json_field('SecurityToken')
    expiration: Optional[str] = json_field('Expiration')
    expiration_at_millis: Optional[int] = json_field('ExpirationAt', default=None)

    @staticmethod
    def parse_expiration_to_millis(expiration: str) -> int:
        pass

    @classmethod
    def from_options(cls, options: Options) -> Optional['DLFToken']:
        pass

class DLFTokenLoader(ABC):

    @abstractmethod
    def load_token(self) -> DLFToken:
        pass

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

class HTTPClient:
    def __init__(self, connect_timeout: int = 180, read_timeout: int = 180,
                 max_retries: int = 3):
        pass

    def get(self, url: str, **kwargs) -> requests.Response:
        pass

class DLFECSTokenLoader(DLFTokenLoader):
    def __init__(self, ecs_metadata_url: str, role_name: Optional[str] = None):
        pass

    def load_token(self) -> DLFToken:
        pass

    def description(self) -> str:
        pass

class DLFTokenLoaderFactory:
    @staticmethod
    def create_token_loader(options: Options) -> Optional['DLFTokenLoader']:
        pass

Import

from pypaimon.api.token_loader import (
    DLFToken, DLFTokenLoader, DLFECSTokenLoader,
    DLFTokenLoaderFactory, HTTPClient
)

I/O Contract

Inputs

Name	Type	Required	Description
access_key_id	str	Yes	DLF access key identifier
access_key_secret	str	Yes	DLF access key secret
security_token	str	No	Temporary security token
expiration	str	No	ISO 8601 expiration timestamp
ecs_metadata_url	str	Yes	ECS metadata service endpoint URL
role_name	str	No	IAM role name (auto-detected if not provided)
options	Options	Yes	Catalog options for factory creation

Outputs

Name	Type	Description
token	DLFToken	DLF authentication token with credentials
loader	DLFTokenLoader	Token loader implementation
expiration_millis	int	Token expiration in milliseconds since epoch

Usage Examples

from pypaimon.api.token_loader import (
    DLFToken, DLFECSTokenLoader, DLFTokenLoaderFactory
)
from pypaimon.common.options import Options

# Create static token
token = DLFToken(
    access_key_id="LTAI4G...",
    access_key_secret="5rT9Y...",
    security_token="CAIS...",
    expiration="2026-02-09T12:00:00Z"
)
print(f"Token expires at: {token.expiration_at_millis} ms")

# Load token from ECS metadata service
loader = DLFECSTokenLoader(
    ecs_metadata_url="http://100.100.100.200/latest/meta-data/Ram/security-credentials/",
    role_name="my-ecs-role"
)
token = loader.load_token()
print(f"Loaded token for key: {token.access_key_id}")

# Auto-detect role name
loader = DLFECSTokenLoader(
    ecs_metadata_url="http://100.100.100.200/latest/meta-data/Ram/security-credentials/"
)
token = loader.load_token()  # Automatically fetches and uses role name

# Create token from catalog options
options = Options({
    'dlf.access.key.id': 'LTAI4G...',
    'dlf.access.key.secret': '5rT9Y...',
    'dlf.access.security.token': 'CAIS...'
})
token = DLFToken.from_options(options)

# Create loader from options using factory
options = Options({
    'dlf.token.loader': 'ecs',
    'dlf.token.ecs.metadata.url': 'http://100.100.100.200/latest/meta-data/Ram/security-credentials/',
    'dlf.token.ecs.role.name': 'my-role'
})
loader = DLFTokenLoaderFactory.create_token_loader(options)
if loader:
    token = loader.load_token()

# Parse expiration timestamp
millis = DLFToken.parse_expiration_to_millis("2026-02-09T12:30:45Z")
print(f"Expiration: {millis} milliseconds since epoch")

Related Pages

• Principle:Apache_Paimon_Authentication