Refactor GreptimeDB Python SDK to separate gRPC and PostgreSQL backends

.gitignore

@@ -18,4 +18,4 @@ test/
 
 /packages/grpc/target/
 .DS_Store
-/docs
+/docs/*

AGENTS.md

@@ -19,7 +19,7 @@ Python write client for GreptimeDB. Package name `greptimedb-ingester`, import `
 ## Build and release
 
 - Root `uv build` builds the pure Python facade. Build the native dependency from `packages/grpc` with maturin. Keep facade/plugin versions, the exact grpc extra pin and lock files synchronized with `scripts/set-version.py`.
-- Test base, postgres-only, gRPC-only and combined installed wheels. Default root Client prefers installed gRPC, otherwise postgres; explicit `.grpc.Client` and `.postgres.Client` select a backend.
+- Test base, postgres-only, gRPC-only and combined installed wheels. The root exports only shared schema and GreptimeError. Import clients explicitly from `.grpc` or `.postgres`; do not add automatic backend selection or legacy import shims.
 - `v*` tags publish the native dependency first, then the public facade, using `PYPI_TOKEN` with access to both distributions. Both backend integrations must pass.
 
 - Wheels are built on the target OS: Linux x86_64, Linux aarch64, Windows x64, macOS Apple Silicon. Do not cross-compile those wheels.

README.md

@@ -35,7 +35,7 @@ print(affected)
 
 `insert` creates missing tables and writes COPY batches (default: 10,000 rows).
 Batches commit separately. The server must support `COPY FROM STDIN`.
-See the [API reference](docs/api.md#postgresql-copy-后端) for supported types and timestamp rules.
+See the [PG getting started guide](docs/postgres-getting-started.en.md) and [COPY example](examples/postgres_example.py).
 
 ## gRPC
 
@@ -47,8 +47,7 @@ client = Client(["127.0.0.1:4001"], database="public")
 ```
 
 Supports `insert`, `delete`, and `bulk_writer`. See [usage](docs/usage.md) and [examples](examples).
-With both backends installed, the root `Client` defaults to gRPC; use the explicit
-imports above to select a backend.
+Import `Client` from the backend you use. Import shared schema types from `greptimedb_ingester`.
 
 ## Development
 
@@ -58,4 +57,4 @@ uv run --extra postgres pytest
 ```
 
 For both backends, use `--all-extras` instead of `--extra postgres` (requires Rust 1.85+).
-See [testing and release](docs/testing.md) for integration tests and publishing.
+See the [CI workflow](.github/workflows/wheels.yml) for integration tests and publishing.

README.zh-CN.md

@@ -35,7 +35,7 @@ print(affected)
 
 `insert` 自动创建不存在的表并分批 COPY，默认每批 10,000 行。
 各批独立提交。服务端须支持 `COPY FROM STDIN`。
-支持类型和时间戳规则见 [API](docs/api.md#postgresql-copy-后端)。
+完整用法见 [PG 入门指南](docs/postgres-getting-started.md)和 [COPY 示例](examples/postgres_example.py)。
 
 ## gRPC
 
@@ -47,8 +47,7 @@ client = Client(["127.0.0.1:4001"], database="public")
 ```
 
 支持 `insert`、`delete` 和 `bulk_writer`，详见 [使用指南](docs/usage.md) 和 [示例](examples)。
-两种后端都安装时，顶层 `Client` 默认 gRPC；
-使用上面的显式导入选择后端。
+从对应后端导入 `Client`，公共 schema 类型从 `greptimedb_ingester` 导入。
 
 ## 开发
 
@@ -58,4 +57,4 @@ uv run --extra postgres pytest
 ```
 
 开发两种后端时，将 `--extra postgres` 换成 `--all-extras`，需要 Rust 1.85+。
-集成测试与发布流程见 [测试文档](docs/testing.md)。
+集成测试与发布流程见 [CI 配置](.github/workflows/wheels.yml)。

docs/usage.md

@@ -29,7 +29,7 @@ uv sync --python 3.10 --extra grpc
 ## 连接
 
 ```python
-from greptimedb_ingester import Client
+from greptimedb_ingester.grpc import Client
 
 client = Client(
     ["127.0.0.1:4001"],
@@ -57,7 +57,8 @@ client.health_check()
 ```python
 from datetime import datetime, timezone
 
-from greptimedb_ingester import Client, Column, ColumnDataType, SemanticType
+from greptimedb_ingester import Column, ColumnDataType, SemanticType
+from greptimedb_ingester.grpc import Client
 
 client = Client(["127.0.0.1:4001"], database="public")
 
@@ -111,7 +112,8 @@ client.delete(
 ```python
 from datetime import datetime, timezone
 
-from greptimedb_ingester import Client, Column, ColumnDataType, SemanticType, WriteOptions
+from greptimedb_ingester import Column, ColumnDataType, SemanticType
+from greptimedb_ingester.grpc import Client, WriteOptions
 
 client = Client(["127.0.0.1:4001"], database="public")
 columns = [
@@ -260,8 +262,7 @@ with Client(["127.0.0.1:4003"], batch_size=5000) as client:
     affected = client.insert("copy_data", columns, rows)
 ```
 
-只安装 `[postgres]` 时，也可直接导入顶层 `Client`。两种后端都安装时，
-顶层 `Client` 默认 gRPC，COPY 用显式导入；schema 和 `GreptimeError` 共用。
+两个后端分别从 `greptimedb_ingester.grpc`、`greptimedb_ingester.postgres` 导入 `Client`；公共 schema 和 `GreptimeError` 从顶层导入。
 COPY 只提供 insert、health_check、close 和 with；不提供 bulk_writer、delete 或 gRPC hints。
 支持类型、时间转换、逐批提交和失败行为见 [README](../README.zh-CN.md)
 和 [API](api.md#postgresql-copy-后端)。完整例子见 `examples/postgres_example.py`。

examples/bench_write.py

@@ -7,7 +7,8 @@
 import argparse
 import time
 
-from greptimedb_ingester import Client, Column, ColumnDataType, SemanticType, WriteOptions
+from greptimedb_ingester import Column, ColumnDataType, SemanticType
+from greptimedb_ingester.grpc import Client, WriteOptions
 
 COLUMNS = [
     Column("device_id", ColumnDataType.STRING, SemanticType.TAG),

examples/bulk_example.py

@@ -7,7 +7,7 @@
 
 from datetime import timedelta
 
-from greptimedb_ingester import Client, WriteOptions
+from greptimedb_ingester.grpc import Client, WriteOptions
 
 from ddl import drop_table
 from type_samples import BULK_COLUMNS, BULK_ROW, INSERT_ROW, WHEN

examples/insert_example.py

@@ -8,7 +8,7 @@
 
 from datetime import timedelta
 
-from greptimedb_ingester import Client
+from greptimedb_ingester.grpc import Client
 
 from ddl import drop_table
 from type_samples import INSERT_COLUMNS, INSERT_ROW, WHEN

packages/grpc/src/bulk.rs

@@ -135,8 +135,11 @@ impl BulkWriter {
     }
 
     #[getter]
-    fn columns(&self) -> Vec<Column> {
-        self.columns.clone()
+    fn columns(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
+        self.columns
+            .iter()
+            .map(|column| column.to_python(py))
+            .collect()
     }
 
     fn alloc_rows(&self, py: Python<'_>, capacity: usize) -> PyResult<Py<Rows>> {
@@ -418,11 +421,8 @@ fn table_schema(
             crate::types::SemanticType::Field
                 if column.data_type == crate::types::ColumnDataType::Decimal128 =>
             {
-                schema.add_decimal128_field(
-                    column.name.clone(),
-                    column.precision.expect("precision"),
-                    column.scale.expect("scale"),
-                )
+                let (precision, scale) = column.decimal_params()?;
+                schema.add_decimal128_field(column.name.clone(), precision, scale)
             }
             crate::types::SemanticType::Field => schema.add_field(column.name.clone(), dtype),
         };

packages/grpc/src/cell.rs

@@ -9,7 +9,7 @@ use crate::error::{raise, RemoteError};
 pub struct DecimalParts {
     pub negative: bool,
     pub digits: String,
-    pub exponent: i32,
+    pub exponent: i64,
 }
 
 #[derive(Clone, Debug)]
@@ -102,11 +102,14 @@ pub fn cell_from_py(obj: &Bound<'_, PyAny>) -> PyResult<Cell> {
         || obj.is_instance_of::<PyList>()
         || obj.is_instance_of::<PyTuple>()
     {
+        let kwargs = PyDict::new(py);
+        kwargs.set_item("allow_nan", false)?;
         let text: String = py
             .import("json")?
             .getattr("dumps")?
-            .call1((obj,))?
-            .extract()?;
+            .call((obj,), Some(&kwargs))
+            .and_then(|value| value.extract())
+            .map_err(|error| raise(format!("invalid JSON: {error}"), false))?;
         return Ok(Cell::Json(text));
     }
 
@@ -131,7 +134,7 @@ fn decimal_cell(obj: &Bound<'_, PyAny>) -> PyResult<Cell> {
     if text.is_empty() {
         text.push('0');
     }
-    let exponent: i32 = parts
+    let exponent: i64 = parts
         .getattr("exponent")?
         .extract()
         .map_err(|_| raise("Decimal exponent is not an integer", false))?;
@@ -145,7 +148,7 @@ fn decimal_cell(obj: &Bound<'_, PyAny>) -> PyResult<Cell> {
 fn timestamp_cell(obj: &Bound<'_, PyAny>, datetime_mod: &Bound<'_, PyAny>) -> PyResult<Cell> {
     let py = obj.py();
     let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
-    let aware = if obj.getattr("tzinfo")?.is_none() {
+    let aware = if obj.getattr("tzinfo")?.is_none() || obj.call_method0("utcoffset")?.is_none() {
         let kwargs = PyDict::new(py);
         kwargs.set_item("tzinfo", utc.clone())?;
         obj.call_method("replace", (), Some(&kwargs))?

packages/grpc/src/client.rs

@@ -12,7 +12,7 @@ use crate::bulk::BulkWriter;
 use crate::convert::{align, build_delete, build_insert, parse_row};
 use crate::error::{raise, RemoteError};
 use crate::runtime;
-use crate::types::{parse_grpc_compression, Column, WriteOptions};
+use crate::types::{checked_duration, parse_grpc_compression, Column, SemanticType, WriteOptions};
 
 #[pyclass]
 #[derive(Clone)]
@@ -117,7 +117,7 @@ impl Client {
         if table.is_empty() {
             return Err(raise("table name must not be empty", false));
         }
-        let aligned = align_rows(&columns, rows)?;
+        let aligned = align_rows(&columns, rows, "insert")?;
         let hints = hints.unwrap_or_default();
         self.execute(py, move |database| async move {
             let request = build_insert(&table, &columns, &aligned)?;
@@ -147,7 +147,16 @@ impl Client {
         if table.is_empty() {
             return Err(raise("table name must not be empty", false));
         }
-        let aligned = align_rows(&columns, rows)?;
+        if columns
+            .iter()
+            .any(|column| column.semantic_type == SemanticType::Field)
+        {
+            return Err(raise(
+                "delete columns must be TAG or TIMESTAMP, not FIELD",
+                false,
+            ));
+        }
+        let aligned = align_rows(&columns, rows, "delete")?;
         self.execute(py, move |database| async move {
             let request = build_delete(&table, &columns, &aligned)?;
             database.delete(request).await.map_err(RemoteError::from)
@@ -212,27 +221,30 @@ impl Client {
 fn align_rows(
     columns: &[Column],
     rows: &Bound<'_, PyAny>,
+    operation: &str,
 ) -> PyResult<Vec<Vec<crate::cell::Cell>>> {
     if columns.is_empty() {
-        return Err(raise("insert requires at least one column", false));
+        return Err(raise(
+            format!("{operation} requires at least one column"),
+            false,
+        ));
     }
     let mut aligned = Vec::new();
     for row in rows.try_iter()? {
         let input = parse_row(&row?)?;
         aligned.push(align(columns, &input)?);
     }
     if aligned.is_empty() {
-        return Err(raise("cannot insert an empty row batch", false));
+        return Err(raise(
+            format!("cannot {operation} an empty row batch"),
+            false,
+        ));
     }
     Ok(aligned)
 }
 
 fn optional_duration(value: Option<f64>, name: &str) -> PyResult<Option<Duration>> {
-    match value {
-        None => Ok(None),
-        Some(secs) if secs >= 0.0 => Ok(Some(Duration::from_secs_f64(secs))),
-        Some(_) => Err(raise(format!("{name} must be >= 0"), false)),
-    }
+    value.map(|secs| checked_duration(secs, name)).transpose()
 }
 
 fn build_ingester_client(