This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 9cfeda78 [python] Add stub test to catch pyi drift in CI (#616)
9cfeda78 is described below
commit 9cfeda7876929257d0bf1d68911560c971350fbc
Author: Kaiqi Dong <[email protected]>
AuthorDate: Wed Jun 10 23:59:15 2026 +0200
[python] Add stub test to catch pyi drift in CI (#616)
* stub test to catch pyi drift
* update development
---
.github/workflows/build_and_test_python.yml | 37 ++++++++++
bindings/python/DEVELOPMENT.md | 19 ++++++
bindings/python/fluss/__init__.pyi | 100 +++++++++++++++++++---------
bindings/python/stubtest-allowlist.txt | 27 ++++++++
4 files changed, 150 insertions(+), 33 deletions(-)
diff --git a/.github/workflows/build_and_test_python.yml
b/.github/workflows/build_and_test_python.yml
index a8d3f05b..52c4d274 100644
--- a/.github/workflows/build_and_test_python.yml
+++ b/.github/workflows/build_and_test_python.yml
@@ -98,3 +98,40 @@ jobs:
path: cluster-logs/
if-no-files-found: ignore
retention-days: 3
+
+ stubtest:
+ # For stub tests, we only needs the module importable, so no
+ # test cluster is started and a single Python version should be sufficient.
+ timeout-minutes: 30
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v6
+
+ - name: Set up Python
+ uses: actions/setup-python@v6
+ with:
+ python-version: "3.12"
+
+ - name: Install uv
+ uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78
+
+ - name: Install protoc
+ uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b #
v3.0.0
+ with:
+ repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+ - name: Rust Cache
+ uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 #
v2.9.1
+
+ - name: Build Python bindings
+ working-directory: bindings/python
+ run: |
+ uv sync --extra dev --no-install-project
+ uv run --no-sync maturin develop --uv
+
+ - name: Check stub drift (stubtest)
+ working-directory: bindings/python
+ run: |
+ uv run --no-sync python -m mypy.stubtest fluss \
+ --mypy-config-file pyproject.toml \
+ --allowlist stubtest-allowlist.txt
diff --git a/bindings/python/DEVELOPMENT.md b/bindings/python/DEVELOPMENT.md
index 65bb37bc..40e473c9 100644
--- a/bindings/python/DEVELOPMENT.md
+++ b/bindings/python/DEVELOPMENT.md
@@ -44,6 +44,25 @@ uv run ruff check python/
uv run mypy python/
```
+## Stub Drift Check
+
+The public API is typed in `fluss/__init__.pyi`. Because that stub is
+hand-written while the runtime classes come from the compiled PyO3 module, the
+two can drift apart. `stubtest` (shipped with `mypy`) compares them and is run
+in CI. After changing the binding's public surface, rebuild the module and run:
+
+```bash
+uv sync --extra dev --no-install-project
+uv run --no-sync maturin develop --uv
+uv run --no-sync python -m mypy.stubtest fluss \
+ --mypy-config-file pyproject.toml \
+ --allowlist stubtest-allowlist.txt
+```
+
+`stubtest-allowlist.txt` holds the few differences that are inherent to PyO3
and
+cannot be expressed in the stub. Keep it minimal — stubtest treats an unused
+allowlist entry as an error, so stale lines must be removed.
+
## Run Examples
Each example is standalone and runnable on its own. They default to a local
diff --git a/bindings/python/fluss/__init__.pyi
b/bindings/python/fluss/__init__.pyi
index 5a8ecb5d..7d5bfa73 100644
--- a/bindings/python/fluss/__init__.pyi
+++ b/bindings/python/fluss/__init__.pyi
@@ -20,7 +20,6 @@
from enum import IntEnum
from types import TracebackType
from typing import (
- Any,
AsyncIterator,
Dict,
Iterator,
@@ -28,12 +27,14 @@ from typing import (
Optional,
Tuple,
Union,
+ final,
overload,
)
import pandas as pd
import pyarrow as pa
+@final
class ChangeType(IntEnum):
"""Represents the type of change for a record in a log."""
@@ -52,6 +53,7 @@ class ChangeType(IntEnum):
"""Returns a short string representation (+A, +I, -U, +U, -D)."""
...
+@final
class ScanRecord:
"""Represents a single scan record with metadata.
@@ -78,6 +80,7 @@ class ScanRecord:
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
+@final
class RecordBatch:
"""Represents a batch of records with metadata."""
@@ -100,6 +103,7 @@ class RecordBatch:
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
+@final
class ScanRecords:
"""A collection of scan records grouped by bucket.
@@ -130,18 +134,19 @@ class ScanRecords:
...
def __len__(self) -> int: ...
@overload
- def __getitem__(self, index: int) -> ScanRecord: ...
+ def __getitem__(self, index: int, /) -> ScanRecord: ...
@overload
- def __getitem__(self, index: slice) -> List[ScanRecord]: ...
+ def __getitem__(self, index: slice, /) -> List[ScanRecord]: ...
@overload
- def __getitem__(self, bucket: TableBucket) -> List[ScanRecord]: ...
- def __contains__(self, bucket: TableBucket) -> bool: ...
+ def __getitem__(self, bucket: TableBucket, /) -> List[ScanRecord]: ...
+ def __contains__(self, bucket: TableBucket, /) -> bool: ...
def __iter__(self) -> Iterator[ScanRecord]: ...
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
+@final
class Config:
- def __init__(self, properties: Optional[Dict[str, str]] = None) -> None:
...
+ def __new__(cls, properties: Optional[Dict[str, str]] = None) -> Config:
...
@property
def bootstrap_servers(self) -> str: ...
@bootstrap_servers.setter
@@ -247,6 +252,7 @@ class Config:
@security_sasl_password.setter
def security_sasl_password(self, password: str) -> None: ...
+@final
class FlussConnection:
@staticmethod
async def create(config: Config) -> FlussConnection: ...
@@ -256,19 +262,20 @@ class FlussConnection:
def __enter__(self) -> FlussConnection: ...
def __exit__(
self,
- exc_type: Optional[type],
- exc_value: Optional[BaseException],
- traceback: Optional[TracebackType],
+ exc_type: Optional[type] = None,
+ exc_value: Optional[BaseException] = None,
+ traceback: Optional[TracebackType] = None,
) -> bool: ...
async def __aenter__(self) -> FlussConnection: ...
async def __aexit__(
self,
- exc_type: Optional[type],
- exc_value: Optional[BaseException],
- traceback: Optional[TracebackType],
+ exc_type: Optional[type] = None,
+ exc_value: Optional[BaseException] = None,
+ traceback: Optional[TracebackType] = None,
) -> bool: ...
def __repr__(self) -> str: ...
+@final
class ServerNode:
"""Information about a server node in the Fluss cluster."""
@@ -294,6 +301,7 @@ class ServerNode:
...
def __repr__(self) -> str: ...
+@final
class FlussAdmin:
async def create_database(
self,
@@ -426,20 +434,22 @@ class FlussAdmin:
def __repr__(self) -> str: ...
+@final
class DatabaseDescriptor:
"""Descriptor for a Fluss database (comment and custom properties)."""
- def __init__(
- self,
+ def __new__(
+ cls,
comment: Optional[str] = None,
custom_properties: Optional[Dict[str, str]] = None,
- ) -> None: ...
+ ) -> DatabaseDescriptor: ...
@property
def comment(self) -> Optional[str]: ...
def get_custom_properties(self) -> Dict[str, str]: ...
def __repr__(self) -> str: ...
+@final
class DatabaseInfo:
"""Information about a Fluss database."""
@@ -452,6 +462,7 @@ class DatabaseInfo:
def modified_time(self) -> int: ...
def __repr__(self) -> str: ...
+@final
class TableScan:
"""Builder for creating log scanners with flexible configuration.
@@ -514,6 +525,7 @@ class TableScan:
...
def __repr__(self) -> str: ...
+@final
class FlussTable:
def new_scan(self) -> TableScan:
"""Create a new table scan builder for configuring and creating log
scanners.
@@ -545,6 +557,7 @@ class FlussTable:
def has_primary_key(self) -> bool: ...
def __repr__(self) -> str: ...
+@final
class TableAppend:
"""Builder for creating an AppendWriter.
@@ -557,6 +570,7 @@ class TableAppend:
def create_writer(self) -> AppendWriter: ...
def __repr__(self) -> str: ...
+@final
class TableUpsert:
"""Builder for creating an UpsertWriter, with optional partial update.
@@ -580,6 +594,7 @@ class TableUpsert:
def create_writer(self) -> UpsertWriter: ...
def __repr__(self) -> str: ...
+@final
class TableLookup:
"""Builder for creating a Lookuper or PrefixLookuper.
@@ -608,6 +623,7 @@ class TableLookup:
...
def __repr__(self) -> str: ...
+@final
class TablePrefixLookup:
"""Builder for creating a PrefixLookuper.
@@ -620,6 +636,7 @@ class TablePrefixLookup:
def create_lookuper(self) -> "PrefixLookuper": ...
def __repr__(self) -> str: ...
+@final
class AppendWriter:
def append(self, row: dict | list | tuple) -> WriteResultHandle:
"""Append a single row to the table.
@@ -674,9 +691,9 @@ class AppendWriter:
...
async def __aexit__(
self,
- exc_type: Optional[type],
- exc_value: Optional[BaseException],
- traceback: Optional[TracebackType],
+ exc_type: Optional[type] = None,
+ exc_value: Optional[BaseException] = None,
+ traceback: Optional[TracebackType] = None,
) -> bool:
"""
Exit the async context manager.
@@ -687,6 +704,7 @@ class AppendWriter:
...
def __repr__(self) -> str: ...
+@final
class UpsertWriter:
"""Writer for upserting and deleting data in a Fluss primary key table."""
@@ -728,9 +746,9 @@ class UpsertWriter:
...
async def __aexit__(
self,
- exc_type: Optional[type],
- exc_value: Optional[BaseException],
- traceback: Optional[TracebackType],
+ exc_type: Optional[type] = None,
+ exc_value: Optional[BaseException] = None,
+ traceback: Optional[TracebackType] = None,
) -> bool:
"""
Exit the async context manager.
@@ -742,6 +760,7 @@ class UpsertWriter:
def __repr__(self) -> str: ...
+@final
class WriteResultHandle:
"""Handle for a pending write (append/upsert/delete). Ignore for
fire-and-forget, or await handle.wait() for ack."""
@@ -751,6 +770,7 @@ class WriteResultHandle:
def __repr__(self) -> str: ...
+@final
class Lookuper:
"""Lookuper for performing primary key lookups on a Fluss table."""
@@ -767,6 +787,7 @@ class Lookuper:
...
def __repr__(self) -> str: ...
+@final
class PrefixLookuper:
"""Lookuper for performing prefix key lookups on a Fluss table.
@@ -790,6 +811,7 @@ class PrefixLookuper:
...
def __repr__(self) -> str: ...
+@final
class LogScanner:
"""Scanner for reading log data from a Fluss table.
@@ -954,19 +976,23 @@ class LogScanner:
def __repr__(self) -> str: ...
def __aiter__(self) -> AsyncIterator[Union[ScanRecord, RecordBatch]]: ...
+@final
class Schema:
- def __init__(
- self, schema: pa.Schema, primary_keys: Optional[List[str]] = None
- ) -> None: ...
+ def __new__(
+ cls, schema: pa.Schema, primary_keys: Optional[List[str]] = None
+ ) -> Schema: ...
def get_column_names(self) -> List[str]: ...
def get_column_types(self) -> List[str]: ...
def get_columns(self) -> List[Tuple[str, str]]: ...
def get_primary_keys(self) -> List[str]: ...
def __str__(self) -> str: ...
+@final
class TableDescriptor:
- def __init__(
- self,
+ # Runtime accepts ``schema`` plus ``**kwargs``; the keyword-only parameters
+ # below document the supported options for type checkers and IDEs.
+ def __new__(
+ cls,
schema: Schema,
*,
partition_keys: Optional[List[str]] = None,
@@ -977,11 +1003,12 @@ class TableDescriptor:
kv_format: Optional[str] = None,
properties: Optional[Dict[str, str]] = None,
custom_properties: Optional[Dict[str, str]] = None,
- ) -> None: ...
+ ) -> TableDescriptor: ...
def get_schema(self) -> Schema: ...
+@final
class TablePath:
- def __init__(self, database: str, table: str) -> None: ...
+ def __new__(cls, database_name: str, table_name: str) -> TablePath: ...
@property
def database_name(self) -> str: ...
@property
@@ -990,8 +1017,9 @@ class TablePath:
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
def __hash__(self) -> int: ...
- def __eq__(self, other: object) -> bool: ...
+ def __eq__(self, other: object, /) -> bool: ...
+@final
class TableInfo:
@property
def table_id(self) -> int: ...
@@ -1018,16 +1046,18 @@ class TableInfo:
def get_column_names(self) -> List[str]: ...
def get_column_count(self) -> int: ...
+@final
class FlussError(Exception):
message: str
error_code: int
- def __init__(self, message: str, error_code: int = -2) -> None: ...
+ def __new__(cls, message: str, error_code: int = -2) -> FlussError: ...
def __str__(self) -> str: ...
@property
def is_retriable(self) -> bool: ...
+@final
class LakeSnapshot:
- def __init__(self, snapshot_id: int) -> None: ...
+ def __new__(cls, snapshot_id: int) -> LakeSnapshot: ...
@property
def snapshot_id(self) -> int: ...
@property
@@ -1037,8 +1067,9 @@ class LakeSnapshot:
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
+@final
class TableBucket:
- def __init__(self, table_id: int, bucket: int) -> None: ...
+ def __new__(cls, table_id: int, bucket: int) -> TableBucket: ...
@staticmethod
def with_partition(
table_id: int, partition_id: int, bucket: int
@@ -1050,10 +1081,11 @@ class TableBucket:
@property
def partition_id(self) -> Optional[int]: ...
def __hash__(self) -> int: ...
- def __eq__(self, other: object) -> bool: ...
+ def __eq__(self, other: object, /) -> bool: ...
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
+@final
class PartitionInfo:
"""Information about a partition."""
@@ -1067,6 +1099,7 @@ class PartitionInfo:
...
def __repr__(self) -> str: ...
+@final
class ErrorCode:
"""Named constants for Fluss API error codes.
@@ -1137,6 +1170,7 @@ class ErrorCode:
INVALID_ALTER_TABLE_EXCEPTION: int
DELETION_DISABLED_EXCEPTION: int
+@final
class OffsetSpec:
"""Offset specification for list_offsets(), matching Java's OffsetSpec.
diff --git a/bindings/python/stubtest-allowlist.txt
b/bindings/python/stubtest-allowlist.txt
new file mode 100644
index 00000000..19de24dd
--- /dev/null
+++ b/bindings/python/stubtest-allowlist.txt
@@ -0,0 +1,27 @@
+# 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.
+
+# Allowlist for `python -m mypy.stubtest fluss`. Each entry is a regex matched
+# against the fully-qualified symbol and suppresses a stub/runtime difference
+# that is inherent to PyO3 and cannot be expressed in the stub. Keep this list
+# minimal: stubtest errors on unused entries, so a stale line fails CI.
+
+# PyO3 cannot embed a literal default value in the generated text signature, so
+# the real `error_code=-2` default is introspected as `Ellipsis`. We keep the
+# accurate `-2` default in the stub for documentation and ignore the resulting
+# default-value mismatch on the constructor.
+fluss.FlussError.__new__
