Repository: tonbo-io/tonbo (language: python)

Issue:
Panic when DB get a removed key in MemTable
`DB::remove` will insert `None`to mutable table, and it will return an `Entry` with `None` value. And [`DB::get`](https://github.com/tonbo-io/tonbo/blob/93623053bbce6c0663c86489ba2630bce2d4288f/src/lib.rs#L350) receives a function that contains a `TransactionEntry` which only provides a `get` method.

However when `TransactionEntry::get` invoked, it will result in unwrap on None err

https://github.com/tonbo-io/tonbo/blob/93623053bbce6c0663c86489ba2630bce2d4288f/src/transaction.rs#L224


Discussion / hints:

Fix patch:
diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml
new file mode 100644
index 00000000..42c9478d
--- /dev/null
+++ b/bindings/python/Cargo.toml
@@ -0,0 +1,30 @@
+[package]
+name = "tonbo-python"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = ["cdylib"]
+
+[workspace]
+
+[dependencies]
+fusio = { git = "https://github.com/tonbo-io/fusio.git", package = "fusio", version = "0.2.0", features = [
+    "aws",
+    "tokio",
+] }
+fusio-dispatch = { git = "https://github.com/tonbo-io/fusio.git", package = "fusio-dispatch", version = "0.2.0", features = [
+    "aws",
+    "tokio",
+] }
+futures = { version = "0.3" }
+pyo3 = { version = "0.21.2", features = [
+    "extension-module",
+    "abi3",
+    "abi3-py310",
+] }
+pyo3-asyncio = { package = "pyo3-asyncio-0-21", version = "0.21", features = [
+    "tokio-runtime",
+] }
+tokio = { version = "1", features = ["rt-multi-thread"] }
+tonbo = { version = ">=0", path = "../../" }
diff --git a/bindings/python/README.md b/bindings/python/README.md
new file mode 100644
index 00000000..0c3a9dc7
--- /dev/null
+++ b/bindings/python/README.md
@@ -0,0 +1,84 @@
+# Tonbo Python Binding
+
+This package intends to build a native python binding for [Tonbo](https://github.com/tonbo-io/tonbo).
+
+Tonbo's Python bindings can be used to build data-intensive applications, including other types of databases.
+
+## Example
+
+```py
+from tonbo import DbOption, Column, DataType, Record, TonboDB, Bound
+from tonbo.fs import from_filesystem_path
+import asyncio
+
+# define a Tonbo record
+@Record
+class User:
+    age = Column(DataType.INT8, name="age", primary_key=True)
+    height = Column(DataType.INT16, name="height", nullable=True)
+    weight = Column(DataType.INT8, name="weight", nullable=False)
+
+async def main():
+
+    db = TonboDB(DbOption(from_filesystem_path("./db_path/users")), User())
+
+    await db.insert(User(age=18, height=175, weight=60))
+    record = await db.get(18)
+    assert record == {"age": 18, "height": 175, "weight": 60}
+
+    txn = await db.transaction()
+    txn.insert(User(age=19, height=195, weight=75))
+    result = await txn.get(19)
+    assert result == {"age": 19, "height": 195, "weight": 75}
+
+    # commit transaction
+    await txn.commit()
+
+    txn = await db.transaction()
+    # range scan, supports pushing down and limit
+    scan = await txn.scan(
+        Bound.Excluded(18), None, limit=100, projection=["age", "weight"]
+    )
+    async for record in scan:
+        print(record)
+
+asyncio.run(main())
+```
+
+See [examples](example/README.md) for more information.
+
+### Roadmap
+
+- Remote storage API mapping and test
+- Integrate with other Arrow analytical tools
+
+## Development
+
+This assumes that you have Rust and cargo installed. We use the [pyo3](https://github.com/PyO3/pyo3) to generate a native Python module and use [maturin](https://github.com/PyO3/maturin) to build Rust-based Python packages.
+
+First, follow the commands below to build a new Python virtualenv, and install maturin into the virtualenv using Python's package manager, pip:
+
+```bash
+# setup virtualenv
+python -m venv .env
+# activate venv
+source .env/bin/activate
+
+# install maturin
+pip install maturin
+# build bindings
+maturin develop
+```
+
+Whenever Rust code changes run:
+
+```bash
+maturin develop
+```
+
+Run tests:
+
+```bash
+maturin develop -E test
+python -m pytest
+```
diff --git a/bindings/python/example/README.md b/bindings/python/example/README.md
new file mode 100644
index 00000000..2754d23c
--- /dev/null
+++ b/bindings/python/example/README.md
@@ -0,0 +1,15 @@
+# Example
+
+Here are some examples that can help you use Python binding of [Tonbo](https://github.com/tonbo-io/tonbo)
+
+### Declaration
+
+- The [record.py](record.py) file shows how to define a `Tonbo` record
+- The [write_batch.py](write_batch.py) file shows how to insert records in batch
+- The [transaction.py](transaction.py) file shows how to use transaction
+- The [declare.py](declare.py) file shows how to use `Tonbo` for simple data operations.
+- The [multiple_instance.py](multiple_instance.py) file shows how to combine different `Tonbo` record with `TonboDB`s.
+
+### Fusion storage
+
+The [fusion_storage.py](fusion_storage.py) file shows how to store data across different storage.
diff --git a/bindings/python/example/declare.py b/bindings/python/example/declare.py
new file mode 100644
index 00000000..24e12a71
--- /dev/null
+++ b/bindings/python/example/declare.py
@@ -0,0 +1,71 @@
+from tonbo import DbOption, Column, DataType, Record, TonboDB, Bound
+import asyncio
+import tempfile
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    age = Column(DataType.Int16, name="age", nullable=True)
+    name = Column(DataType.String, name="name", nullable=False)
+    email = Column(DataType.String, name="email", nullable=True)
+    data = Column(DataType.Bytes, name="data", nullable=True)
+
+
+async def main():
+    temp_dir = tempfile.TemporaryDirectory()
+
+    db = TonboDB(DbOption(temp_dir.name), User())
+    await db.insert(User(id=18, age=175, name="Alice"))
+
+    record = await db.get(18)
+    assert record == {
+        "id": 18,
+        "age": 175,
+        "name": "Alice",
+        "email": None,
+        "data": None,
+    }
+
+    txn = await db.transaction()
+    result = await txn.get(18)
+    assert result == {
+        "id": 18,
+        "age": 175,
+        "name": "Alice",
+        "email": None,
+        "data": None,
+    }
+
+    txn.insert(
+        User(
+            id=19,
+            age=195,
+            name="Bob",
+            data=b"Hello Tonbo!",
+            email="contact@tonbo.io",
+        )
+    )
+    result = await txn.get(19)
+    assert result == {
+        "id": 19,
+        "age": 195,
+        "name": "Bob",
+        "email": "contact@tonbo.io",
+        "data": b"Hello Tonbo!",
+    }
+
+    await txn.commit()
+    txn = await db.transaction()
+    scan = await txn.scan(
+        Bound.Excluded(18),
+        None,
+        limit=100,
+        projection=["id", "email", "data"],
+    )
+    async for record in scan:
+        assert record["age"] is None
+        print(record)
+    await txn.commit()
+
+
+asyncio.run(main())
diff --git a/bindings/python/example/fusion_storage.py b/bindings/python/example/fusion_storage.py
new file mode 100644
index 00000000..87b8f079
--- /dev/null
+++ b/bindings/python/example/fusion_storage.py
@@ -0,0 +1,57 @@
+import asyncio
+import os
+
+from tonbo import DbOption, Column, DataType, Record, TonboDB
+from tonbo.fs import from_filesystem_path, FsOptions
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    name = Column(DataType.String, name="name")
+    email = Column(DataType.String, name="email", nullable=True)
+    age = Column(DataType.UInt8, name="age")
+    data = Column(DataType.Bytes, name="data")
+
+
+async def main():
+    if not os.path.exists("db_path/user/l0"):
+        os.makedirs("db_path/user/l0")
+    if not os.path.exists("db_path/user/l1"):
+        os.makedirs("db_path/user/l1")
+
+    option = DbOption(from_filesystem_path("db_path/user"))
+    option.level_path(0, from_filesystem_path("db_path/user/l0"), FsOptions.Local())
+    option.level_path(1, from_filesystem_path("db_path/user/l1"), FsOptions.Local())
+
+    option.immutable_chunk_num = 1
+    option.major_threshold_with_sst_size = 3
+    option.level_sst_magnification = 1
+    option.max_sst_file_size = 1 * 1024
+
+    db = TonboDB(option, User())
+    for i in range(0, 1000):
+        if i % 50 == 0:
+            await db.flush()
+        await db.insert(
+            User(
+                id=i,
+                age=i % 128,
+                name=str(i * 10),
+                email=str(i * 20),
+                data=b"Hello Tonbo!",
+            )
+        )
+
+    for i in range(0, 1000):
+        user = await db.get(i)
+        assert user == {
+            "id": i,
+            "name": str(i * 10),
+            "email": str(i * 20),
+            "age": i % 128,
+            "data": b"Hello Tonbo!",
+        }
+
+
+asyncio.run(main())
diff --git a/bindings/python/example/multiple_instance.py b/bindings/python/example/multiple_instance.py
new file mode 100644
index 00000000..23f53a84
--- /dev/null
+++ b/bindings/python/example/multiple_instance.py
@@ -0,0 +1,71 @@
+import asyncio
+import tempfile
+
+from tonbo import Column, DataType, TonboDB, RecordBatch, DbOption, Record
+from tonbo.fs import from_filesystem_path
+
+
+@Record
+class User:
+    id = Column(DataType.UInt64, name="id", primary_key=True)
+    age = Column(datatype=DataType.UInt8, name="age", nullable=True)
+    name = Column(DataType.String, name="name")
+    email = Column(DataType.String, name="email")
+
+
+@Record
+class Pet:
+    id = Column(DataType.UInt64, name="id", primary_key=True)
+    type = Column(DataType.String, name="type")
+    name = Column(DataType.String, name="name")
+    master = Column(datatype=DataType.UInt64, name="master", nullable=True)
+
+
+@Record
+class FoundRecord:
+    user_id = Column(DataType.UInt64, name="user_id", primary_key=True)
+    pet_id = Column(DataType.UInt64, name="pet_id")
+    email = Column(DataType.String, name="email")
+    status = Column(DataType.Boolean, name="status")
+
+
+async def main():
+    user_temp_dir = tempfile.TemporaryDirectory()
+    pet_temp_dir = tempfile.TemporaryDirectory()
+    found_temp_dir = tempfile.TemporaryDirectory()
+    user_db = TonboDB(DbOption(from_filesystem_path(user_temp_dir.name)), User())
+    course_db = TonboDB(DbOption(from_filesystem_path(pet_temp_dir.name)), Pet())
+    found_db = TonboDB(DbOption(from_filesystem_path(found_temp_dir.name)), FoundRecord())
+
+    user_txn = await user_db.transaction()
+    for i in range(0, 100):
+        user_txn.insert(User(id=i, age=i, name=str(i * 20), email=str(i * 100)))
+    user_txn.commit()
+
+    pet_txn = await course_db.transaction()
+    pet_txn.insert(Pet(id=1, type="dog", name="Hachi", master=10))
+    pet_txn.insert(Pet(id=2, type="cat", name="Kitty", master=21))
+    pet_txn.insert(Pet(id=3, type="dragonfly", name="Tonbo", master=11))
+
+    await pet_txn.commit()
+
+    user_txn = await user_db.transaction()
+    found_txn = await found_db.transaction()
+    user_scan = await user_txn.scan(lower=None, high=None, projection=["id", "name", "email"])
+
+    async for user in user_scan:
+        pet_txn = await course_db.transaction()
+        pet_scan = await pet_txn.scan(lower=None, high=None, projection=["master", "name"])
+        async for pet in pet_scan:
+            if user["id"] == pet["master"]:
+                found_txn.insert(FoundRecord(user_id=user["id"], pet_id=pet["id"], email=user["email"], status=False))
+
+    await found_txn.commit()
+
+    found_txn = await found_db.transaction()
+    found_scan = await found_txn.scan(lower=None, high=None)
+    async for found in found_scan:
+        print(found)
+
+
+asyncio.run(main())
diff --git a/bindings/python/example/record.py b/bindings/python/example/record.py
new file mode 100644
index 00000000..9ebff2d4
--- /dev/null
+++ b/bindings/python/example/record.py
@@ -0,0 +1,14 @@
+from tonbo import Record, Column, DataType
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    age = Column(datatype=DataType.UInt8, name="age", nullable=True)
+    # nullable is `False` by default
+    name = Column(DataType.String, name="name")
+    height = Column(DataType.Int16, name="height")
+    enabled = Column(DataType.Boolean, name="enabled")
+    email = Column(DataType.String, name="email", nullable=True)
+    data = Column(DataType.Bytes, name="data", nullable=True)
+
diff --git a/bindings/python/example/transaction.py b/bindings/python/example/transaction.py
new file mode 100644
index 00000000..574bc051
--- /dev/null
+++ b/bindings/python/example/transaction.py
@@ -0,0 +1,58 @@
+from tonbo import DbOption, Column, DataType, Record, TonboDB, Bound
+import asyncio
+import tempfile
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    height = Column(DataType.Int16, name="height", nullable=True)
+    name = Column(DataType.String, name="name", nullable=False)
+    email = Column(DataType.String, name="email", nullable=True)
+    data = Column(DataType.Bytes, name="data", nullable=True)
+
+
+async def main():
+    temp_dir = tempfile.TemporaryDirectory()
+
+    db = TonboDB(DbOption(temp_dir.name), User())
+    # create a new transaction
+    txn = await db.transaction()
+
+    # insert with class
+    txn.insert(
+        User(
+            id=19,
+            height=195,
+            name="Bob",
+            data=b"Hello Tonbo!",
+            email="contact@tonbo.io",
+        )
+    )
+    result = await txn.get(19)
+    assert result == {
+        "id": 19,
+        "height": 195,
+        "name": "Bob",
+        "email": "contact@tonbo.io",
+        "data": b"Hello Tonbo!",
+    }
+
+    # commit a transaction
+    await txn.commit()
+
+    txn = await db.transaction()
+    # support push down limit, filter and projection
+    scan = await txn.scan(
+        Bound.Excluded(18),
+        None,
+        limit=100,
+        projection=["id", "email", "data"],
+    )
+    async for record in scan:
+        assert record["height"] is None
+        print(record)
+    await txn.commit()
+
+
+asyncio.run(main())
diff --git a/bindings/python/example/write_batch.py b/bindings/python/example/write_batch.py
new file mode 100644
index 00000000..13a651ad
--- /dev/null
+++ b/bindings/python/example/write_batch.py
@@ -0,0 +1,32 @@
+import asyncio
+import tempfile
+
+from tonbo import Column, DataType, TonboDB, RecordBatch, DbOption, Record
+from tonbo.fs import from_filesystem_path
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    age = Column(datatype=DataType.UInt8, name="age", nullable=True)
+    name = Column(DataType.String, name="name")
+    email = Column(DataType.String, name="email", nullable=True)
+    data = Column(DataType.Bytes, name="data", nullable=True)
+
+
+async def main():
+    temp_dir = tempfile.TemporaryDirectory()
+    db = TonboDB(DbOption(from_filesystem_path(temp_dir.name)), User())
+    batch = RecordBatch()
+    for i in range(0, 100):
+        # you must add record to `RecordBatch` one by one
+        batch.append(User(id= i, age=i, name=str(i * 20)))
+
+    await db.insert_batch(batch)
+
+    for i in range(0, 100):
+        user = await db.get(i)
+        assert user == { "id": i, "age": i, "name": str(i * 20), "email": None, "data": None }
+
+
+asyncio.run(main())
diff --git a/bindings/python/pyproject.toml b/bindings/python/pyproject.toml
new file mode 100644
index 00000000..795ca0de
--- /dev/null
+++ b/bindings/python/pyproject.toml
@@ -0,0 +1,24 @@
+[build-system]
+requires = ["maturin>=1.7,<2.0"]
+build-backend = "maturin"
+
+[project]
+name = "tonbo"
+readme = "README.md"
+requires-python = ">=3.10"
+classifiers = [
+    "Programming Language :: Rust",
+    "Programming Language :: Python :: Implementation :: CPython",
+    "Programming Language :: Python :: Implementation :: PyPy",
+]
+dynamic = ["version"]
+
+[project.optional-dependencies]
+test = ["pytest", "pytest-asyncio"]
+docs = ["pdoc"]
+
+[tool.maturin]
+features = ["pyo3/extension-module"]
+
+module-name = "tonbo._tonbo"
+python-source = "python"
diff --git a/bindings/python/python/tonbo/__init__.py b/bindings/python/python/tonbo/__init__.py
new file mode 100644
index 00000000..96bc598d
--- /dev/null
+++ b/bindings/python/python/tonbo/__init__.py
@@ -0,0 +1,4 @@
+from ._tonbo import *
+
+__doc__ = _tonbo.__doc__
+__all__ = _tonbo.__all__
diff --git a/bindings/python/python/tonbo/__init__.pyi b/bindings/python/python/tonbo/__init__.pyi
new file mode 100644
index 00000000..4fb79a66
--- /dev/null
+++ b/bindings/python/python/tonbo/__init__.pyi
@@ -0,0 +1,160 @@
+from typing import Any, AsyncIterable, final
+from enum import Enum, auto
+from tonbo import error as error
+from tonbo.fs import FsOptions
+
+@final
+class Record: ...
+
+@final
+class Bound(Enum):
+    """Tonbo range for scan. None for unbounded"""
+
+    Included = auto()
+    Excluded = auto()
+
+@final
+class DataType(Enum):
+    """Tonbo data type."""
+
+    UInt8 = auto()
+    UInt16 = auto()
+    UInt32 = auto()
+    UInt64 = auto()
+    Int8 = auto()
+    Int16 = auto()
+    Int32 = auto()
+    Int64 = auto()
+    String = auto()
+    Boolean = auto()
+    Bytes = auto()
+
+@final
+class Column:
+    """Column represents properties of a field."""
+
+    name: str
+    datatype: DataType
+    nullable: bool
+    primary_key: bool
+    def __init__(
+        self,
+        datatype: DataType,
+        name: str,
+        nullable: bool = False,
+        primary_key: bool = False,
+    ) -> None: ...
+
+@final
+class RecordBatch:
+    """Column represents properties of a field."""
+    def __init__(self) -> None: ...
+    def append(self, record: object) -> None: ...
+
+class DbOption:
+    """Tonbo configurations."""
+
+    clean_channel_buffer: int
+    immutable_chunk_num: int
+    level_sst_magnification: int
+    major_default_oldest_table_num: int
+    major_threshold_with_sst_size: int
+    max_sst_file_size: int
+    version_log_snapshot_threshold: int
+    use_wal: bool
+    path: str
+
+    def __init__(self, path: str) -> None:
+        """Create a new :py:class:`DbOption` with the given path. Note: the path must exist"""
+        ...
+
+    def level_path(self, level: int, path: str, fs_options: FsOptions) -> None:
+        """Set path for assigned level
+
+        Args:
+            level: Level for output.
+            path: Path for output
+            fs_options: Local or S3
+        """
+        ...
+
+@final
+class Transaction:
+    """Tonbo transaction."""
+
+    async def get(
+            self, key: Any, projection: list[str] = ["*"]
+    ) -> dict[str, Any] | None:
+        """Get record from db.
+
+        Args:
+            key: Primary key of record.
+            projection: fields to projection
+        """
+        ...
+    def insert(self, record: object) -> None:
+        """Insert record to db."""
+        ...
+    def remove(self, key: Any) -> None:
+        """Remove record from db.
+        Args:
+            key: Primary key of record.
+        """
+        ...
+    async def scan(
+            self,
+            lower: Bound | None,
+            high: Bound | None,
+            limit: int | None,
+            projection: list[str] = ["*"],
+    ) -> AsyncIterable[dict[str, Any]]:
+        """Create an async stream for scanning.
+
+        Args:
+            lower: Lower bound of range. Use None represent unbounded.
+            high: High bound of range. Use None represent unbounded.
+            limit: max number records to scan
+            projection: fields to projection
+        """
+        ...
+    async def commit(self) -> None:
+        """Commit :py:class:`Transaction`."""
+        ...
+
+@final
+class TonboDB:
+    def __init__(self, option: DbOption, schema: Any) -> None:
+        """Create a new :py:class:`TonboDB` with the given configuration options.
+
+        Args:
+            option: Configuration options.
+            schema: Schema of record.
+        """
+        ...
+    async def get(self, key: Any) -> dict[str, Any] | None:
+        """Get record from db.
+
+        Args:
+            key: Primary key of record.
+        """
+        ...
+    async def insert(self, record: object) -> None:
+        """Insert a record to db."""
+        ...
+    async def insert_batch(self, record_batch: RecordBatch) -> None:
+        """Insert :py:class:`RecordBatch` to db."""
+        ...
+    async def remove(self, key: Any) -> None:
+        """Remove record from db.
+
+        Args:
+            key: Primary key of record.
+        """
+        ...
+    async def transaction(self) -> Transaction:
+        """Create a new :py:class:`Transaction`."""
+        ...
+    async def flush(self) -> None:
+        """Try to execute compaction."""
+        ...
+
diff --git a/bindings/python/python/tonbo/error.pyi b/bindings/python/python/tonbo/error.pyi
new file mode 100644
index 00000000..71f00b90
--- /dev/null
+++ b/bindings/python/python/tonbo/error.pyi
@@ -0,0 +1,31 @@
+class CommitError(Exception):
+    """Commit error"""
+    pass
+
+class DbError(Exception):
+    """DB error"""
+    pass
+
+class DecodeError(Exception):
+    """Decode error"""
+    pass
+
+class RecoverError(Exception):
+    """Recover error"""
+    pass
+
+class ExceedsMaxLevelError(Exception):
+    """Exceeds max level"""
+    pass
+
+class WriteConflictError(Exception):
+    """Write conflict"""
+    pass
+
+class RepeatedCommitError(Exception):
+    """Repeated commit the same transaction"""
+    pass
+
+class PathParseError(Exception):
+    """Parse path error"""
+    pass
\ No newline at end of file
diff --git a/bindings/python/python/tonbo/fs.pyi b/bindings/python/python/tonbo/fs.pyi
new file mode 100644
index 00000000..0700d76e
--- /dev/null
+++ b/bindings/python/python/tonbo/fs.pyi
@@ -0,0 +1,43 @@
+from typing import final
+
+@final
+class AwsCredential:
+    key_id: str
+    secret_key: str
+    token: str | None
+
+    def __init__(
+        self, key_id: str, secret_key: str, token: str | None = None
+    ) -> None: ...
+
+@final
+class FsOptions:
+    class Local: ...
+
+    class S3:
+        bucket: str
+        credential: AwsCredential | None
+        region: str | None
+        sign_payload: bool | None
+        checksum: bool | None
+
+        def __init__(
+            self,
+            bucket: str,
+            credential: AwsCredential | None = None,
+            region: str | None = None,
+            sign_payload: bool | None = None,
+            checksum: bool | None = None,
+        ) -> None: ...
+
+def parse(path: str) -> str: ...
+
+def from_filesystem_path(path: str) -> str:
+    """Parse path to a filesystem relative path"""
+    ...
+def from_absolute_path(path: str) -> str:
+    """Parse path to a filesystem absolute path"""
+    ...
+def from_url_path(path: str) -> str:
+    """Parse an url path"""
+    ...
diff --git a/bindings/python/src/column.rs b/bindings/python/src/column.rs
new file mode 100644
index 00000000..ada350af
--- /dev/null
+++ b/bindings/python/src/column.rs
@@ -0,0 +1,72 @@
+use std::{
+    any::Any,
+    fmt::{Display, Formatter},
+    sync::Arc,
+};
+
+use pyo3::{pyclass, pymethods};
+use tonbo::record::{ColumnDesc, Datatype};
+
+use crate::datatype::DataType;
+
+#[pyclass]
+#[derive(Clone)]
+pub struct Column {
+    pub name: String,
+    pub datatype: DataType,
+    pub nullable: bool,
+    pub primary_key: bool,
+    pub(crate) value: Arc<dyn Any>,
+}
+
+unsafe impl Send for Column {}
+unsafe impl Sync for Column {}
+
+impl Column {}
+
+#[pymethods]
+impl Column {
+    #[new]
+    #[pyo3(signature= (datatype, name, nullable=false, primary_key=false))]
+    pub fn new(datatype: DataType, name: String, nullable: bool, primary_key: bool) -> Self {
+        if primary_key && nullable {
+            panic!("Primary key should not be nullable!")
+        }
+        let value = datatype.none_value();
+        Self {
+            name,
+            datatype,
+            nullable,
+            primary_key,
+            value,
+        }
+    }
+
+    fn __str__(&self) -> String {
+        format!("{}", &self)
+    }
+}
+
+impl Display for Column {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Column")
+            .field("name", &self.name)
+            .field("type", &self.datatype)
+            .field("nullable", &self.nullable)
+            .field("primary_key", &self.primary_key)
+            .finish()
+    }
+}
+
+impl From<Column> for ColumnDesc {
+    fn from(col: Column) -> Self {
+        let datatype = Datatype::from(col.datatype);
+        ColumnDesc::new(col.name, datatype, col.nullable)
+    }
+}
+impl From<Column> for tonbo::record::Column {
+    fn from(col: Column) -> Self {
+        let datatype = Datatype::from(col.datatype);
+        tonbo::record::Column::new(datatype, col.name, col.value, col.nullable)
+    }
+}
diff --git a/bindings/python/src/datatype.rs b/bindings/python/src/datatype.rs
new file mode 100644
index 00000000..41030eab
--- /dev/null
+++ b/bindings/python/src/datatype.rs
@@ -0,0 +1,112 @@
+use std::{
+    any::Any,
+    fmt::{Debug, Formatter},
+    sync::Arc,
+};
+
+use pyo3::{pyclass, PyObject, Python, ToPyObject};
+use tonbo::record::Datatype;
+
+#[pyclass]
+#[derive(PartialEq, Clone)]
+pub enum DataType {
+    UInt8,
+    UInt16,
+    UInt32,
+    UInt64,
+    Int8,
+    Int16,
+    Int32,
+    Int64,
+    String,
+    Boolean,
+    Bytes,
+}
+
+impl Debug for DataType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DataType::UInt8 => f.write_str("u8"),
+            DataType::UInt16 => f.write_str("u16"),
+            DataType::UInt32 => f.write_str("u32"),
+            DataType::UInt64 => f.write_str("u64"),
+            DataType::Int8 => f.write_str("i8"),
+            DataType::Int16 => f.write_str("i16"),
+            DataType::Int32 => f.write_str("i32"),
+            DataType::Int64 => f.write_str("i64"),
+            DataType::String => f.write_str("str"),
+            DataType::Boolean => f.write_str("bool"),
+            DataType::Bytes => f.write_str("bytes"),
+        }
+    }
+}
+
+impl DataType {
+    #[allow(unused)]
+    pub(crate) fn default_value(&self, py: Python<'_>) -> PyObject {
+        match self {
+            DataType::UInt8 => u8::default().to_object(py),
+            DataType::UInt16 => u16::default().to_object(py),
+            DataType::UInt32 => u32::default().to_object(py),
+            DataType::UInt64 => u64::default().to_object(py),
+            DataType::Int8 => i8::default().to_object(py),
+            DataType::Int16 => i16::default().to_object(py),
+            DataType::Int32 => i32::default().to_object(py),
+            DataType::Int64 => i64::default().to_object(py),
+            DataType::String => String::default().to_object(py),
+            DataType::Boolean => bool::default().to_object(py),
+            DataType::Bytes => Vec::<u8>::default().to_object(py),
+        }
+    }
+    pub(crate) fn none_value(&self) -> Arc<dyn Any> {
+        match self {
+            DataType::UInt8 => Arc::new(Option::<u8>::None),
+            DataType::UInt16 => Arc::new(Option::<u16>::None),
+            DataType::UInt32 => Arc::new(Option::<u32>::None),
+            DataType::UInt64 => Arc::new(Option::<u64>::None),
+            DataType::Int8 => Arc::new(Option::<i8>::None),
+            DataType::Int16 => Arc::new(Option::<i16>::None),
+            DataType::Int32 => Arc::new(Option::<i32>::None),
+            DataType::Int64 => Arc::new(Option::<i64>::None),
+            DataType::String => Arc::new(Option::<String>::None),
+            DataType::Boolean => Arc::new(Option::<bool>::None),
+            DataType::Bytes => Arc::new(Option::<Vec<u8>>::None),
+        }
+    }
+}
+
+impl From<DataType> for Datatype {
+    fn from(datatype: DataType) -> Self {
+        match datatype {
+            DataType::UInt8 => Datatype::UInt8,
+            DataType::UInt16 => Datatype::UInt16,
+            DataType::UInt32 => Datatype::UInt32,
+            DataType::UInt64 => Datatype::UInt64,
+            DataType::Int8 => Datatype::Int8,
+            DataType::Int16 => Datatype::Int16,
+            DataType::Int32 => Datatype::Int32,
+            DataType::Int64 => Datatype::Int64,
+            DataType::String => Datatype::String,
+            DataType::Boolean => Datatype::Boolean,
+            DataType::Bytes => Datatype::Bytes,
+        }
+    }
+}
+
+impl From<&DataType> for Datatype {
+    fn from(datatype: &DataType) -> Self {
+        match datatype {
+            DataType::UInt8 => Datatype::UInt8,
+            DataType::UInt16 => Datatype::UInt16,
+            DataType::UInt32 => Datatype::UInt32,
+            DataType::UInt64 => Datatype::UInt64,
+            DataType::Int8 => Datatype::Int8,
+            DataType::Int16 => Datatype::Int16,
+            DataType::Int32 => Datatype::Int32,
+            DataType::Int64 => Datatype::Int64,
+            DataType::String => Datatype::String,
+            DataType::Boolean => Datatype::Boolean,
+            DataType::Bytes => Datatype::Bytes,
+        }
+    }
+}
diff --git a/bindings/python/src/db.rs b/bindings/python/src/db.rs
new file mode 100644
index 00000000..c99aea3f
--- /dev/null
+++ b/bindings/python/src/db.rs
@@ -0,0 +1,169 @@
+use std::sync::Arc;
+
+use pyo3::{
+    prelude::*,
+    pyclass, pymethods,
+    types::{PyDict, PyMapping},
+    Py, PyAny, PyResult, Python,
+};
+use pyo3_asyncio::tokio::{future_into_py, get_runtime};
+use tonbo::{
+    executor::tokio::TokioExecutor,
+    record::{ColumnDesc, DynRecord},
+    DB,
+};
+
+use crate::{
+    column::Column,
+    error::CommitError,
+    options::DbOption,
+    record_batch::RecordBatch,
+    transaction::Transaction,
+    utils::{to_col, to_dict},
+};
+
+type PyExecutor = TokioExecutor;
+
+#[pyclass]
+pub struct TonboDB {
+    desc: Arc<Vec<Column>>,
+    primary_key_index: usize,
+    db: Arc<DB<DynRecord, PyExecutor>>,
+}
+
+#[pymethods]
+impl TonboDB {
+    #[new]
+    fn new(py: Python<'_>, option: DbOption, record: Py<PyAny>) -> PyResult<Self> {
+        let dict = record.getattr(py, "__dict__")?;
+        let values = dict.downcast_bound::<PyMapping>(py)?.values()?;
+        let mut desc = vec![];
+        let mut cols = vec![];
+        let mut primary_key_index = None;
+        let mut primary_key_name = None;
+
+        for i in 0..values.len()? {
+            let value = values.get_item(i)?;
+            if let Ok(bound_col) = value.downcast::<Column>() {
+                let col = bound_col.extract::<Column>()?;
+                if col.primary_key {
+                    if primary_key_index.is_some() {
+                        panic!("Multiple primary keys is not allowed!")
+                    }
+                    primary_key_index = Some(desc.len());
+                    primary_key_name = Some(col.name.clone());
+                }
+                cols.push(col.clone());
+                desc.push(ColumnDesc::from(col));
+            }
+        }
+        let option = option.into_option(primary_key_index.unwrap(), primary_key_name.unwrap());
+        let db = get_runtime()
+            .block_on(async {
+                DB::with_schema(
+                    option,
+                    TokioExecutor::new(),
+                    desc,
+                    primary_key_index.unwrap(),
+                )
+                .await
+            })
+            .unwrap();
+        Ok(Self {
+            db: Arc::new(db),
+            desc: Arc::new(cols),
+            primary_key_index: primary_key_index.expect("Primary key not found"),
+        })
+    }
+
+    /// Insert record to `TonboDB`.
+    ///
+    /// * `record`: Primary key of record that is to be removed
+    fn insert<'py>(&'py self, py: Python<'py>, record: Py<PyAny>) -> PyResult<Bound<PyAny>> {
+        let mut cols = vec![];
+        let dict = record.getattr(py, "__dict__")?;
+        let values = dict.downcast_bound::<PyMapping>(py)?.values()?;
+
+        for i in 0..values.len()? {
+            let value = values.get_item(i)?;
+            if let Ok(bound_col) = value.downcast::<Column>() {
+                let col = tonbo::record::Column::from(bound_col.extract::<Column>()?);
+                cols.push(col);
+            }
+        }
+
+        let db = self.db.clone();
+        let primary_key_index = self.primary_key_index;
+
+        future_into_py(py, async move {
+            db.insert(DynRecord::new(cols, primary_key_index))
+                .await
+                .map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| PyDict::new_bound(py).into_py(py)))
+        })
+    }
+
+    fn insert_batch<'py>(&'py self, py: Python<'py>, batch: RecordBatch) -> PyResult<Bound<PyAny>> {
+        let record_batch = batch.into_record_batch();
+        let db = self.db.clone();
+
+        future_into_py(py, async move {
+            db.insert_batch(record_batch.into_iter())
+                .await
+                .map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| PyDict::new_bound(py).into_py(py)))
+        })
+    }
+
+    /// Get record from `TonboDB`.
+    ///
+    /// * `key`: Primary key of record
+    fn get<'py>(&'py self, py: Python<'py>, key: Py<PyAny>) -> PyResult<Bound<'py, PyAny>> {
+        let col_desc = self.desc.get(self.primary_key_index).unwrap();
+        let col = to_col(py, col_desc, key);
+        let db = self.db.clone();
+        let primary_key_index = self.primary_key_index;
+        future_into_py(py, async move {
+            let record = db
+                .get(&col, |e| Some(e.get().columns))
+                .await
+                .map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| match record {
+                Some(record) => to_dict(py, primary_key_index, record).into_py(py),
+                None => py.None(),
+            }))
+        })
+    }
+
+    /// Remove record from `TonboDB`.
+    ///
+    /// * `key`: Primary key of record that is to be removed
+    fn remove<'py>(&'py self, py: Python<'py>, key: Py<PyAny>) -> PyResult<Bound<PyAny>> {
+        let col_desc = self.desc.get(self.primary_key_index).unwrap();
+        let col = to_col(py, col_desc, key);
+        let db = self.db.clone();
+        future_into_py(py, async move {
+            let ret = db.remove(col).await.map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| ret.into_py(py)))
+        })
+    }
+
+    /// Open an optimistic ACID `Transaction`.
+    fn transaction<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<PyAny>> {
+        let db = self.db.clone();
+        let desc = self.desc.clone();
+        future_into_py(py, async move {
+            let txn = db.transaction().await;
+            Ok(Transaction::new(txn, desc.clone()))
+        })
+    }
+
+    fn flush<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<PyAny>> {
+        let db = self.db.clone();
+
+        future_into_py(py, async move {
+            db.flush().await.map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| py.None()))
+        })
+    }
+}
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
new file mode 100644
index 00000000..8e91373a
--- /dev/null
+++ b/bindings/python/src/error.rs
@@ -0,0 +1,85 @@
+use pyo3::{
+    create_exception,
+    exceptions::{PyException, PyIOError, PyValueError},
+    pyclass, PyErr,
+};
+use tonbo::record::DynRecord;
+create_exception!(tonbo, DecodeError, PyException, "Decode exception");
+create_exception!(tonbo, RecoverError, PyException, "Recover exception");
+create_exception!(
+    tonbo,
+    ExceedsMaxLevelError,
+    PyException,
+    "Exceeds max level exception"
+);
+
+create_exception!(
+    tonbo,
+    WriteConflictError,
+    PyException,
+    "Write conflict exception"
+);
+create_exception!(tonbo, InnerError, PyException, "Inner exception");
+
+create_exception!(
+    tonbo,
+    RepeatedCommitError,
+    PyException,
+    "Repeated commit exception"
+);
+
+create_exception!(tonbo, PathParseError, PyException, "Path parse exception");
+
+pub(crate) fn repeated_commit_err() -> PyErr {
+    RepeatedCommitError::new_err("Transaction has been committed!")
+}
+
+#[pyclass]
+pub(crate) struct DbError(tonbo::DbError<DynRecord>);
+
+#[pyclass]
+pub(crate) struct CommitError(tonbo::transaction::CommitError<DynRecord>);
+
+impl From<DbError> for PyErr {
+    fn from(err: DbError) -> Self {
+        match err.0 {
+            tonbo::DbError::Io(err) => PyIOError::new_err(err.to_string()),
+            tonbo::DbError::Version(err) => PyValueError::new_err(err.to_string()),
+            tonbo::DbError::Parquet(err) => InnerError::new_err(err.to_string()),
+            tonbo::DbError::UlidDecode(err) => DecodeError::new_err(err.to_string()),
+            tonbo::DbError::Fusio(err) => InnerError::new_err(err.to_string()),
+            tonbo::DbError::Recover(err) => RecoverError::new_err(err.to_string()),
+            tonbo::DbError::WalWrite(err) => PyIOError::new_err(err.to_string()),
+            tonbo::DbError::ExceedsMaxLevel => ExceedsMaxLevelError::new_err("Exceeds max level"),
+        }
+    }
+}
+
+impl From<CommitError> for PyErr {
+    fn from(err: CommitError) -> Self {
+        match err.0 {
+            tonbo::transaction::CommitError::Io(err) => PyIOError::new_err(err.to_string()),
+            tonbo::transaction::CommitError::Parquet(err) => InnerError::new_err(err.to_string()),
+            tonbo::transaction::CommitError::Database(err) => DbError::from(err).into(),
+            tonbo::transaction::CommitError::WriteConflict(key) => {
+                WriteConflictError::new_err(key.name)
+            }
+            tonbo::transaction::CommitError::SendCompactTaskError(err) => {
+                InnerError::new_err(err.to_string())
+            }
+            tonbo::transaction::CommitError::ChannelClose => InnerError::new_err("channel close"),
+        }
+    }
+}
+
+impl From<tonbo::DbError<DynRecord>> for DbError {
+    fn from(err: tonbo::DbError<DynRecord>) -> Self {
+        DbError(err)
+    }
+}
+
+impl From<tonbo::transaction::CommitError<DynRecord>> for CommitError {
+    fn from(err: tonbo::transaction::CommitError<DynRecord>) -> Self {
+        CommitError(err)
+    }
+}
diff --git a/bindings/python/src/fs.rs b/bindings/python/src/fs.rs
new file mode 100644
index 00000000..b9b7f730
--- /dev/null
+++ b/bindings/python/src/fs.rs
@@ -0,0 +1,94 @@
+use fusio::path::Path;
+use pyo3::{pyclass, pyfunction, pymethods, types::PyString, Bound, PyResult, Python};
+
+use crate::PathParseError;
+
+#[pyclass(get_all, set_all)]
+#[derive(Debug, Clone)]
+pub struct AwsCredential {
+    pub key_id: String,
+    pub secret_key: String,
+    pub token: Option<String>,
+}
+
+impl From<AwsCredential> for fusio::remotes::aws::AwsCredential {
+    fn from(cred: AwsCredential) -> Self {
+        fusio::remotes::aws::AwsCredential {
+            key_id: cred.key_id,
+            secret_key: cred.secret_key,
+            token: cred.token,
+        }
+    }
+}
+
+#[pymethods]
+impl AwsCredential {
+    #[new]
+    fn new(key_id: String, secret_key: String, token: Option<String>) -> Self {
+        Self {
+            key_id,
+            secret_key,
+            token,
+        }
+    }
+}
+
+#[pyclass]
+#[derive(Debug, Clone)]
+pub enum FsOptions {
+    Local {},
+    S3 {
+        bucket: String,
+        credential: Option<AwsCredential>,
+        region: Option<String>,
+        sign_payload: Option<bool>,
+        checksum: Option<bool>,
+    },
+}
+
+impl From<FsOptions> for fusio_dispatch::FsOptions {
+    fn from(opt: FsOptions) -> Self {
+        match opt {
+            FsOptions::Local {} => fusio_dispatch::FsOptions::Local,
+            FsOptions::S3 {
+                bucket,
+                credential,
+                region,
+                sign_payload,
+                checksum,
+            } => fusio_dispatch::FsOptions::S3 {
+                bucket,
+                credential: credential.map(fusio::remotes::aws::AwsCredential::from),
+                region,
+                sign_payload,
+                checksum,
+            },
+        }
+    }
+}
+
+#[pyfunction]
+pub fn parse(path: String, py: Python) -> PyResult<Bound<PyString>> {
+    let path = Path::parse(path).map_err(|e| PathParseError::new_err(e.to_string()))?;
+    Ok(PyString::new_bound(py, path.as_ref()))
+}
+
+#[pyfunction]
+pub fn from_filesystem_path(path: String, py: Python) -> PyResult<Bound<PyString>> {
+    let path =
+        Path::from_filesystem_path(path).map_err(|e| PathParseError::new_err(e.to_string()))?;
+    Ok(PyString::new_bound(py, path.as_ref()))
+}
+
+#[pyfunction]
+pub fn from_absolute_path(path: String, py: Python) -> PyResult<Bound<PyString>> {
+    let path =
+        Path::from_absolute_path(path).map_err(|e| PathParseError::new_err(e.to_string()))?;
+    Ok(PyString::new_bound(py, path.as_ref()))
+}
+
+#[pyfunction]
+pub fn from_url_path(path: String, py: Python) -> PyResult<Bound<PyString>> {
+    let path = Path::from_url_path(path).map_err(|e| PathParseError::new_err(e.to_string()))?;
+    Ok(PyString::new_bound(py, path.as_ref()))
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
new file mode 100644
index 00000000..4c9e2319
--- /dev/null
+++ b/bindings/python/src/lib.rs
@@ -0,0 +1,112 @@
+use error::*;
+use pyo3::prelude::*;
+use record_batch::RecordBatch;
+
+use crate::record::Record;
+
+mod column;
+mod datatype;
+mod db;
+mod error;
+mod fs;
+mod options;
+mod range;
+mod record;
+mod record_batch;
+mod stream;
+mod transaction;
+mod utils;
+
+pub use column::*;
+pub use datatype::*;
+pub use db::*;
+pub use fs::*;
+pub use options::*;
+pub use stream::*;
+pub use transaction::*;
+
+use crate::error::{DecodeError, WriteConflictError};
+
+/// Tonbo Python binding
+///
+/// ## Usage
+/// ```python
+/// from tonbo import DbOption, Column, DataType, Record, TonboDB, Bound
+/// from tonbo.fs import from_filesystem_path
+/// import asyncio
+///
+/// @Record
+/// class User:
+///    id = Column(DataType.Int64, name="id", primary_key=True)
+///    age = Column(DataType.Int16, name="age", nullable=True)
+///    name = Column(DataType.String, name="name", nullable=False)
+///
+/// async def main():
+///     db = TonboDB(DbOption(from_filesystem_path("db_path/user")), User())
+///     await db.insert(User(id=18, age=175, name="Alice"))
+///     record = await db.get(18)
+///     print(record)
+///
+///     # use transcaction
+///     txn = await db.transaction()
+///     result = await txn.get(18)
+///     scan = await txn.scan(Bound.Included(18), None, limit=10, projection=["id", "name"])
+///
+///     async for record in scan:
+///         print(record)
+///
+/// asyncio.run(main())
+/// ````
+#[pymodule]
+fn _tonbo(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_class::<TonboDB>()?;
+    m.add_class::<DbOption>()?;
+    m.add_class::<DataType>()?;
+    m.add_class::<Column>()?;
+    m.add_class::<Record>()?;
+    m.add_class::<Transaction>()?;
+    m.add_class::<ScanStream>()?;
+    m.add_class::<range::Bound>()?;
+    m.add_class::<RecordBatch>()?;
+
+    let fs_module = PyModule::new_bound(py, "fs")?;
+    fs_module.add_class::<FsOptions>()?;
+    fs_module.add_class::<AwsCredential>()?;
+    fs_module.add_function(wrap_pyfunction!(parse, &fs_module)?)?;
+    fs_module.add_function(wrap_pyfunction!(from_filesystem_path, &fs_module)?)?;
+    fs_module.add_function(wrap_pyfunction!(from_absolute_path, &fs_module)?)?;
+    fs_module.add_function(wrap_pyfunction!(from_url_path, &fs_module)?)?;
+
+    m.add_submodule(&fs_module)?;
+    py.import_bound("sys")?
+        .getattr("modules")?
+        .set_item("tonbo.fs", fs_module)?;
+
+    let error_module = PyModule::new_bound(py, "error")?;
+    error_module.add_class::<DbError>()?;
+    error_module.add_class::<CommitError>()?;
+
+    error_module.add("DecodeError", py.get_type_bound::<DecodeError>())?;
+    error_module.add("RecoverError", py.get_type_bound::<RecoverError>())?;
+    error_module.add(
+        "ExceedsMaxLevelError",
+        py.get_type_bound::<ExceedsMaxLevelError>(),
+    )?;
+    error_module.add(
+        "WriteConflictError",
+        py.get_type_bound::<WriteConflictError>(),
+    )?;
+    error_module.add("InnerError", py.get_type_bound::<InnerError>())?;
+    error_module.add(
+        "RepeatedCommitError",
+        py.get_type_bound::<RepeatedCommitError>(),
+    )?;
+    error_module.add("PathParseError", py.get_type_bound::<PathParseError>())?;
+
+    m.add_submodule(&error_module)?;
+    py.import_bound("sys")?
+        .getattr("modules")?
+        .set_item("tonbo.error", error_module)?;
+
+    Ok(())
+}
diff --git a/bindings/python/src/options.rs b/bindings/python/src/options.rs
new file mode 100644
index 00000000..d1a91c6e
--- /dev/null
+++ b/bindings/python/src/options.rs
@@ -0,0 +1,102 @@
+use fusio::path::Path;
+use pyo3::{pyclass, pymethods, PyResult};
+use tonbo::record::DynRecord;
+
+use crate::{ExceedsMaxLevelError, FsOptions};
+
+pub(crate) const MAX_LEVEL: usize = 7;
+
+/// configure the operating parameters of each component in the `DB`
+#[pyclass]
+#[derive(Debug, Clone)]
+pub struct DbOption {
+    /// cached message size in parquet cleaner
+    #[pyo3(get, set)]
+    clean_channel_buffer: usize,
+    /// len threshold of `immutables` when minor compaction is triggered
+    #[pyo3(get, set)]
+    immutable_chunk_num: usize,
+    /// magnification that triggers major compaction between different levels
+    #[pyo3(get, set)]
+    level_sst_magnification: usize,
+    #[pyo3(get, set)]
+    major_default_oldest_table_num: usize,
+    /// threshold for the number of `parquet` when major compaction is triggered
+    #[pyo3(get, set)]
+    major_threshold_with_sst_size: usize,
+    /// Maximum size of each parquet
+    #[pyo3(get, set)]
+    max_sst_file_size: usize,
+    #[pyo3(get, set)]
+    version_log_snapshot_threshold: u32,
+    #[pyo3(get, set)]
+    use_wal: bool,
+    /// build the `DB` storage directory based on the passed path
+    #[pyo3(get, set)]
+    path: String,
+    #[pyo3(get, set)]
+    base_fs: FsOptions,
+    level_paths: Vec<Option<(String, FsOptions)>>,
+}
+
+#[pymethods]
+impl DbOption {
+    #[new]
+    fn new(path: String) -> Self {
+        Self {
+            clean_channel_buffer: 10,
+            immutable_chunk_num: 3,
+            level_sst_magnification: 10,
+            major_default_oldest_table_num: 3,
+            major_threshold_with_sst_size: 4,
+            max_sst_file_size: 256 * 1024 * 1024,
+            version_log_snapshot_threshold: 200,
+            use_wal: true,
+            path,
+            base_fs: FsOptions::Local {},
+            level_paths: vec![None; MAX_LEVEL],
+        }
+    }
+
+    fn level_path(&mut self, level: usize, path: String, fs_options: FsOptions) -> PyResult<()> {
+        if level >= MAX_LEVEL {
+            ExceedsMaxLevelError::new_err("Exceeds max level");
+        }
+        self.level_paths[level] = Some((path, fs_options));
+        Ok(())
+    }
+}
+
+impl DbOption {
+    pub(crate) fn into_option(
+        self,
+        primary_key_index: usize,
+        primary_key_name: String,
+    ) -> tonbo::DbOption<DynRecord> {
+        let mut opt =
+            tonbo::DbOption::with_path(Path::from(self.path), primary_key_name, primary_key_index)
+                .clean_channel_buffer(self.clean_channel_buffer)
+                .immutable_chunk_num(self.immutable_chunk_num)
+                .level_sst_magnification(self.level_sst_magnification)
+                .major_default_oldest_table_num(self.major_default_oldest_table_num)
+                .major_threshold_with_sst_size(self.major_threshold_with_sst_size)
+                .max_sst_file_size(self.max_sst_file_size)
+                .version_log_snapshot_threshold(self.version_log_snapshot_threshold)
+                .base_fs(fusio_dispatch::FsOptions::from(self.base_fs));
+        for (level, path) in self.level_paths.into_iter().enumerate() {
+            if let Some((path, fs_options)) = path {
+                opt = opt
+                    .level_path(
+                        level,
+                        Path::from(path),
+                        fusio_dispatch::FsOptions::from(fs_options),
+                    )
+                    .unwrap();
+            }
+        }
+        if !self.use_wal {
+            opt = opt.disable_wal()
+        }
+        opt
+    }
+}
diff --git a/bindings/python/src/range.rs b/bindings/python/src/range.rs
new file mode 100644
index 00000000..6ac0904b
--- /dev/null
+++ b/bindings/python/src/range.rs
@@ -0,0 +1,21 @@
+use std::ops;
+
+use pyo3::{pyclass, FromPyObject, Py, PyAny, Python};
+
+use crate::{utils::to_col, Column};
+
+#[pyclass]
+#[derive(FromPyObject)]
+pub enum Bound {
+    Included { key: Py<PyAny> },
+    Excluded { key: Py<PyAny> },
+}
+
+impl Bound {
+    pub(crate) fn to_bound(&self, py: Python, col: &Column) -> ops::Bound<tonbo::record::Column> {
+        match self {
+            Bound::Included { key } => ops::Bound::Included(to_col(py, col, key.clone_ref(py))),
+            Bound::Excluded { key } => ops::Bound::Excluded(to_col(py, col, key.clone_ref(py))),
+        }
+    }
+}
diff --git a/bindings/python/src/record.rs b/bindings/python/src/record.rs
new file mode 100644
index 00000000..ecee7436
--- /dev/null
+++ b/bindings/python/src/record.rs
@@ -0,0 +1,120 @@
+use std::sync::Arc;
+
+use pyo3::{
+    prelude::*,
+    pyclass, pymethods,
+    types::{PyDict, PyMapping, PyString},
+    Bound,
+};
+
+use crate::{column::Column, datatype::DataType};
+
+#[pyclass(subclass)]
+pub struct Record {
+    wraps: Py<PyAny>,
+}
+
+#[pymethods]
+impl Record {
+    #[new]
+    fn new(wraps: Py<PyAny>) -> Self {
+        Self { wraps }
+    }
+
+    #[pyo3(signature = ( **kwargs))]
+    fn __call__(&self, py: Python<'_>, kwargs: Option<&Bound<'_, PyDict>>) -> PyResult<Py<PyAny>> {
+        let dict = self.wraps.getattr(py, "__dict__")?;
+        let mapping_proxy = dict.downcast_bound::<PyMapping>(py).unwrap();
+        if let Some(kwargs) = kwargs {
+            for (key, v) in kwargs.iter() {
+                let attr = key.downcast::<PyString>().unwrap();
+                let col_bound = mapping_proxy.get_item(attr).expect("Unknown attr {attr}");
+                let mut col = col_bound.extract::<Column>().unwrap();
+                match col.datatype {
+                    DataType::UInt8 => {
+                        let value = v.extract::<u8>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::UInt16 => {
+                        let value = v.extract::<u16>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::UInt32 => {
+                        let value = v.extract::<u32>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::UInt64 => {
+                        let value = v.extract::<u64>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Int8 => {
+                        let value = v.extract::<i8>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Int16 => {
+                        let value = v.extract::<i16>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Int32 => {
+                        let value = v.extract::<i32>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Int64 => {
+                        let value = v.extract::<i64>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::String => {
+                        let value = v.extract::<String>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Boolean => {
+                        let value = v.extract::<bool>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                    DataType::Bytes => {
+                        let value = v.extract::<Vec<u8>>()?;
+                        match col.nullable {
+                            true => col.value = Arc::new(Some(value)),
+                            false => col.value = Arc::new(value),
+                        }
+                    }
+                };
+                self.wraps.setattr(py, attr, col).unwrap();
+            }
+        }
+
+        let ret = self.wraps.clone_ref(py).into_any();
+
+        Ok(ret)
+    }
+}
diff --git a/bindings/python/src/record_batch.rs b/bindings/python/src/record_batch.rs
new file mode 100644
index 00000000..0064e445
--- /dev/null
+++ b/bindings/python/src/record_batch.rs
@@ -0,0 +1,81 @@
+use pyo3::{
+    pyclass, pymethods,
+    types::{PyAnyMethods, PyMapping, PyMappingMethods},
+    Py, PyAny, PyResult, Python,
+};
+use tonbo::record::DynRecord;
+
+use crate::Column;
+
+#[derive(Clone)]
+struct Record {
+    columns: Vec<tonbo::record::Column>,
+    primary_key_index: usize,
+}
+
+impl Record {
+    fn new(columns: Vec<tonbo::record::Column>, primary_key_index: usize) -> Self {
+        Self {
+            columns,
+            primary_key_index,
+        }
+    }
+}
+
+impl From<Record> for DynRecord {
+    fn from(value: Record) -> Self {
+        tonbo::record::DynRecord::new(value.columns, value.primary_key_index)
+    }
+}
+
+#[pyclass]
+#[derive(Clone)]
+// #[derive(FromPyObject)]
+pub struct RecordBatch {
+    batch_data: Vec<Record>,
+}
+
+#[pymethods]
+impl RecordBatch {
+    #[new]
+    fn new() -> Self {
+        Self {
+            batch_data: Vec::new(),
+        }
+    }
+
+    fn append(&mut self, py: Python, record: Py<PyAny>) -> PyResult<()> {
+        let mut cols = vec![];
+        let dict = record.getattr(py, "__dict__")?;
+        let values = dict.downcast_bound::<PyMapping>(py)?.values()?;
+        let mut primary_key_index = 0;
+        let mut col_idx = 0;
+
+        for i in 0..values.len()? {
+            let value = values.get_item(i)?;
+            if let Ok(bound_col) = value.downcast::<Column>() {
+                let col = bound_col.extract::<Column>()?;
+                if col.primary_key {
+                    primary_key_index = col_idx;
+                }
+                let col = tonbo::record::Column::from(col);
+                cols.push(col);
+                col_idx += 1;
+            }
+        }
+
+        self.batch_data
+            .push(Record::new(cols.clone(), primary_key_index));
+        Ok(())
+    }
+}
+
+impl RecordBatch {
+    pub(crate) fn into_record_batch(self) -> Vec<DynRecord> {
+        let mut batch = vec![];
+        for record in self.batch_data.into_iter() {
+            batch.push(tonbo::record::DynRecord::from(record))
+        }
+        batch
+    }
+}
diff --git a/bindings/python/src/stream.rs b/bindings/python/src/stream.rs
new file mode 100644
index 00000000..0c66171e
--- /dev/null
+++ b/bindings/python/src/stream.rs
@@ -0,0 +1,52 @@
+use std::{pin::Pin, sync::Arc};
+
+use futures::{Stream, TryStreamExt};
+use pyo3::{
+    exceptions::PyStopAsyncIteration, prelude::*, pyclass, pymethods, IntoPy, PyRef, PyRefMut,
+    PyResult, Python,
+};
+use pyo3_asyncio::tokio::future_into_py;
+use tokio::sync::Mutex;
+use tonbo::{parquet::errors::ParquetError, record::DynRecord, stream};
+
+use crate::utils::to_dict;
+
+type AsyncStream =
+    Pin<Box<dyn Stream<Item = Result<stream::Entry<'static, DynRecord>, ParquetError>> + Send>>;
+
+#[pyclass]
+pub struct ScanStream(Arc<Mutex<AsyncStream>>);
+
+impl ScanStream {
+    pub fn new(
+        stream: impl Stream<Item = Result<stream::Entry<'static, DynRecord>, ParquetError>>
+            + 'static
+            + Sized
+            + Send,
+    ) -> Self {
+        Self(Arc::new(Mutex::new(Box::pin(stream))))
+    }
+}
+
+#[pymethods]
+impl ScanStream {
+    fn __aiter__(slf: PyRef<Self>) -> PyRef<Self> {
+        slf
+    }
+
+    fn __anext__(slf: PyRefMut<Self>, py: Python<'_>) -> PyResult<Option<PyObject>> {
+        let stream: Arc<Mutex<AsyncStream>> = Arc::clone(&slf.0);
+        let fut = future_into_py(py, async move {
+            let mut locked_stream = stream.lock().await;
+            let entry = locked_stream.try_next().await.unwrap();
+            match entry {
+                Some(entry) => Ok(Python::with_gil(|py| match entry.value() {
+                    Some(record) => to_dict(py, record.primary_index, record.columns).into_py(py),
+                    None => py.None(),
+                })),
+                None => Err(PyStopAsyncIteration::new_err("stream exhausted")),
+            }
+        })?;
+        Ok(Some(fut.into()))
+    }
+}
diff --git a/bindings/python/src/transaction.rs b/bindings/python/src/transaction.rs
new file mode 100644
index 00000000..70d535d7
--- /dev/null
+++ b/bindings/python/src/transaction.rs
@@ -0,0 +1,220 @@
+use std::{mem::transmute, sync::Arc};
+
+use pyo3::{
+    pyclass, pymethods,
+    types::{PyAnyMethods, PyMapping, PyMappingMethods, PySequenceMethods, PyTuple},
+    Bound, IntoPy, Py, PyAny, PyResult, Python,
+};
+use pyo3_asyncio::tokio::future_into_py;
+use tonbo::{record::DynRecord, transaction, Projection};
+
+use crate::{
+    column::Column,
+    error::{repeated_commit_err, CommitError, DbError},
+    range,
+    stream::ScanStream,
+    utils::{to_bound, to_col, to_dict},
+};
+
+#[pyclass]
+pub struct Transaction {
+    txn: Option<transaction::Transaction<'static, DynRecord>>,
+    desc: Arc<Vec<Column>>,
+    primary_key_index: usize,
+}
+
+impl Transaction {
+    pub(crate) fn new<'txn>(
+        txn: transaction::Transaction<'txn, DynRecord>,
+        desc: Arc<Vec<Column>>,
+    ) -> Self {
+        let primary_key_index = desc
+            .iter()
+            .enumerate()
+            .find(|(_idx, col)| col.primary_key)
+            .unwrap()
+            .0;
+        Transaction {
+            txn: Some(unsafe {
+                transmute::<
+                    transaction::Transaction<'txn, DynRecord>,
+                    transaction::Transaction<'static, DynRecord>,
+                >(txn)
+            }),
+            desc,
+            primary_key_index,
+        }
+    }
+
+    fn projection(&self, projection: Vec<String>) -> Vec<usize> {
+        match projection.contains(&"*".to_string()) {
+            true => (0..self.desc.len()).collect(),
+            false => self
+                .desc
+                .iter()
+                .enumerate()
+                .filter(|(_idx, col)| projection.contains(&col.name))
+                .map(|(idx, _col)| idx)
+                .collect(),
+        }
+    }
+}
+
+#[pymethods]
+impl Transaction {
+    /// Get record from `TonboDB`.
+    ///
+    /// * `key`: Primary key of record
+    /// * `projection`: Fields to projection, all fields by default.
+    #[pyo3(signature= (key, projection=vec!["*".to_owned()]))]
+    fn get<'py>(
+        &'py self,
+        py: Python<'py>,
+        key: Py<PyAny>,
+        projection: Vec<String>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        if self.txn.is_none() {
+            return Err(repeated_commit_err());
+        }
+
+        let col_desc = self.desc.get(self.primary_key_index).unwrap();
+        let col = to_col(py, col_desc, key);
+        let projection = self.projection(projection);
+
+        let txn = self.txn.as_ref().unwrap();
+        let txn = unsafe {
+            transmute::<
+                &transaction::Transaction<'_, DynRecord>,
+                &'static transaction::Transaction<'_, DynRecord>,
+            >(txn)
+        };
+
+        let primary_key_index = self.primary_key_index;
+
+        future_into_py(py, async move {
+            let entry = txn
+                .get(&col, Projection::Parts(projection))
+                .await
+                .map_err(DbError::from)?;
+
+            Python::with_gil(|py| {
+                Ok(match entry {
+                    Some(entry) => to_dict(py, primary_key_index, entry.get().columns).into_py(py),
+                    None => py.None(),
+                })
+            })
+        })
+    }
+
+    /// Insert record to `TonboDB`.
+    ///
+    /// * `record`: Primary key of record that is to be removed
+    fn insert(&mut self, py: Python, record: Py<PyAny>) -> PyResult<()> {
+        if self.txn.is_none() {
+            return Err(repeated_commit_err());
+        }
+        let mut cols = vec![];
+        let dict = record.getattr(py, "__dict__")?;
+        let mapping_proxy = dict.downcast_bound::<PyMapping>(py)?;
+
+        for x in mapping_proxy.items().iter() {
+            let list = x.to_tuple()?;
+            for x in list {
+                let tuple = x.downcast::<PyTuple>()?;
+                let col = tuple.get_item(1)?;
+                if let Ok(bound_col) = col.downcast::<Column>() {
+                    let col = tonbo::record::Column::from(bound_col.extract::<Column>()?);
+                    cols.push(col);
+                }
+            }
+        }
+        let record = DynRecord::new(cols, self.primary_key_index);
+        self.txn.as_mut().unwrap().insert(record);
+        Ok(())
+    }
+
+    /// Remove record from `TonboDB`.
+    ///
+    /// * `key`: Primary key of record that is to be removed
+    fn remove(&mut self, py: Python, key: Py<PyAny>) -> PyResult<()> {
+        if self.txn.is_none() {
+            return Err(repeated_commit_err());
+        }
+
+        let col_desc = self.desc.get(self.primary_key_index).unwrap();
+        let col = to_col(py, col_desc, key);
+
+        self.txn.as_mut().unwrap().remove(col);
+        Ok(())
+    }
+
+    /// Create an async stream for scanning.
+    ///
+    /// * `lower`: - Lower bound of range. Use None represent unbounded.
+    /// * `high`: - High bound of range. Use None represent unbounded.
+    /// * `limit`: - Max number records to scan.
+    /// * `projection`: - Fields to projection in the record. Projection all by default.
+    #[pyo3(signature= (lower, high, limit=None,  projection=vec!["*".to_string()]))]
+    fn scan<'py>(
+        &'py mut self,
+        py: Python<'py>,
+        lower: Option<Py<range::Bound>>,
+        high: Option<Py<range::Bound>>,
+        limit: Option<usize>,
+        projection: Vec<String>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        if self.txn.is_none() {
+            return Err(repeated_commit_err());
+        }
+        let txn = self.txn.as_ref().unwrap();
+        let txn = unsafe {
+            transmute::<
+                &transaction::Transaction<'_, DynRecord>,
+                &'static transaction::Transaction<'_, DynRecord>,
+            >(txn)
+        };
+        let col_desc = self.desc.get(self.primary_key_index).unwrap();
+        let projection = self.projection(projection);
+
+        let (lower, high) = to_bound(py, col_desc, lower, high);
+
+        future_into_py(py, async move {
+            let mut scan = txn.scan((
+                unsafe {
+                    transmute::<
+                        std::ops::Bound<&tonbo::record::Column>,
+                        std::ops::Bound<&'static tonbo::record::Column>,
+                    >(lower.as_ref())
+                },
+                unsafe {
+                    transmute::<
+                        std::ops::Bound<&tonbo::record::Column>,
+                        std::ops::Bound<&'static tonbo::record::Column>,
+                    >(high.as_ref())
+                },
+            ));
+
+            if let Some(limit) = limit {
+                scan = scan.limit(limit);
+            }
+            scan = scan.projection(projection);
+            let stream = scan.take().await.map_err(DbError::from)?;
+
+            let stream = ScanStream::new(stream);
+
+            Ok(Python::with_gil(|py| stream.into_py(py)))
+        })
+    }
+
+    /// Commit `Transaction`
+    fn commit<'py>(&'py mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        if self.txn.is_none() {
+            return Err(repeated_commit_err());
+        }
+        let txn = self.txn.take();
+        future_into_py(py, async move {
+            txn.unwrap().commit().await.map_err(CommitError::from)?;
+            Ok(Python::with_gil(|py| py.None()))
+        })
+    }
+}
diff --git a/bindings/python/src/utils.rs b/bindings/python/src/utils.rs
new file mode 100644
index 00000000..2ff0e77b
--- /dev/null
+++ b/bindings/python/src/utils.rs
@@ -0,0 +1,206 @@
+use std::{any::Any, sync::Arc};
+
+use pyo3::{
+    types::{PyBytes, PyDict, PyDictMethods},
+    Bound, Py, PyAny, Python,
+};
+use tonbo::record::Datatype;
+
+use crate::{column::Column, datatype::DataType, range};
+
+pub(crate) fn to_dict(
+    py: Python,
+    primary_key_index: usize,
+    record: Vec<tonbo::record::Column>,
+) -> Bound<PyDict> {
+    let dict = PyDict::new_bound(py);
+    for (idx, col) in record.iter().enumerate() {
+        match &col.datatype {
+            Datatype::UInt8 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<u8>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<u8>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::UInt16 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<u16>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<u16>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::UInt32 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<u32>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<u32>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::UInt64 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<u64>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<u64>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Int8 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<i8>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<i8>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Int16 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<i16>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<i16>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Int32 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<i32>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<i32>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Int64 => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<i64>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<i64>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::String => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<String>(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<String>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Boolean => {
+                if idx == primary_key_index {
+                    dict.set_item(
+                        col.name.clone(),
+                        col.value.as_ref().downcast_ref::<bool>().unwrap(),
+                    )
+                    .unwrap();
+                } else {
+                    let value = col.value.as_ref().downcast_ref::<Option<bool>>().unwrap();
+                    dict.set_item(col.name.clone(), value).unwrap();
+                }
+            }
+            Datatype::Bytes => {
+                if idx == primary_key_index {
+                    let value = col.value.as_ref().downcast_ref::<Vec<u8>>().unwrap();
+                    let v = PyBytes::new_bound(py, value);
+                    dict.set_item(col.name.clone(), v).unwrap();
+                } else {
+                    let value = col
+                        .value
+                        .as_ref()
+                        .downcast_ref::<Option<Vec<u8>>>()
+                        .unwrap();
+                    dict.set_item(
+                        col.name.clone(),
+                        value.as_ref().map(|v| PyBytes::new_bound(py, v)),
+                    )
+                    .unwrap();
+                }
+            }
+        }
+    }
+    dict
+}
+
+pub(crate) fn to_key(py: Python, datatype: &DataType, key: Py<PyAny>) -> Arc<dyn Any> {
+    match datatype {
+        DataType::UInt8 => Arc::new(key.extract::<u8>(py).unwrap()) as Arc<dyn Any>,
+        DataType::UInt16 => Arc::new(key.extract::<u16>(py).unwrap()) as Arc<dyn Any>,
+        DataType::UInt32 => Arc::new(key.extract::<u32>(py).unwrap()) as Arc<dyn Any>,
+        DataType::UInt64 => Arc::new(key.extract::<u64>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Int8 => Arc::new(key.extract::<i8>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Int16 => Arc::new(key.extract::<i16>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Int32 => Arc::new(key.extract::<i32>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Int64 => Arc::new(key.extract::<i64>(py).unwrap()) as Arc<dyn Any>,
+        DataType::String => Arc::new(key.extract::<String>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Boolean => Arc::new(key.extract::<bool>(py).unwrap()) as Arc<dyn Any>,
+        DataType::Bytes => Arc::new(key.extract::<Vec<u8>>(py).unwrap()) as Arc<dyn Any>,
+    }
+}
+
+pub(crate) fn to_col(py: Python, col: &Column, key: Py<PyAny>) -> tonbo::record::Column {
+    tonbo::record::Column::new(
+        Datatype::from(&col.datatype),
+        col.name.to_owned(),
+        to_key(py, &col.datatype, key),
+        col.nullable,
+    )
+}
+
+pub(crate) fn to_bound(
+    py: Python,
+    col: &Column,
+    lower: Option<Py<range::Bound>>,
+    high: Option<Py<range::Bound>>,
+) -> (
+    std::ops::Bound<tonbo::record::Column>,
+    std::ops::Bound<tonbo::record::Column>,
+) {
+    let lower = match lower {
+        Some(bound) => bound.get().to_bound(py, col),
+        None => std::ops::Bound::Unbounded,
+    };
+    let high = match high {
+        Some(bound) => bound.get().to_bound(py, col),
+        None => std::ops::Bound::Unbounded,
+    };
+
+    (lower, high)
+}
diff --git a/src/lib.rs b/src/lib.rs
index 3932f55a..446478e2 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -333,7 +333,7 @@ where
     pub async fn get<T>(
         &self,
         key: &R::Key,
-        mut f: impl FnMut(TransactionEntry<'_, R>) -> T,
+        mut f: impl FnMut(TransactionEntry<'_, R>) -> Option<T>,
     ) -> Result<Option<T>, CommitError<R>> {
         Ok(self
             .schema
@@ -347,7 +347,13 @@ where
                 Projection::All,
             )
             .await?
-            .map(|e| f(TransactionEntry::Stream(e))))
+            .and_then(|entry| {
+                if entry.value().is_none() {
+                    None
+                } else {
+                    f(TransactionEntry::Stream(entry))
+                }
+            }))
     }
 
     /// scan records with primary keys in the `range` and process them using closure `f`
@@ -1685,6 +1691,40 @@ pub(crate) mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_get_removed() {
+        let temp_dir = TempDir::new().unwrap();
+
+        let mut option = DbOption::from(Path::from_filesystem_path(temp_dir.path()).unwrap());
+        option.immutable_chunk_num = 1;
+        option.immutable_chunk_max_num = 1;
+        option.major_threshold_with_sst_size = 3;
+        option.major_default_oldest_table_num = 1;
+        option.trigger_type = TriggerType::Length(5);
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new()).await.unwrap();
+
+        for (idx, item) in test_items().into_iter().enumerate() {
+            if idx % 2 == 0 {
+                db.write(item, 0.into()).await.unwrap();
+            } else {
+                db.remove(item.vstring).await.unwrap();
+            }
+        }
+
+        for i in 0..40 {
+            let vstring = db
+                .get(&i.to_string(), |e| Some(e.get().vstring.to_string()))
+                .await
+                .unwrap();
+            if i % 2 == 0 {
+                assert!(vstring.is_some());
+                assert_eq!(vstring, Some(i.to_string()));
+            } else {
+                assert!(vstring.is_none());
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_read_write_dyn() {
         let temp_dir = TempDir::new().unwrap();
diff --git a/src/record/runtime/array.rs b/src/record/runtime/array.rs
index c2c31fe3..a0f193e5 100644
--- a/src/record/runtime/array.rs
+++ b/src/record/runtime/array.rs
@@ -2,11 +2,14 @@ use std::{any::Any, mem, sync::Arc};
 
 use arrow::{
     array::{
-        Array, ArrayBuilder, ArrayRef, BooleanArray, BooleanBufferBuilder, BooleanBuilder,
-        GenericBinaryArray, GenericBinaryBuilder, Int16Array, Int32Array, Int64Array, Int8Array,
-        PrimitiveBuilder, StringArray, StringBuilder, UInt32Builder,
+        Array, ArrayBuilder, ArrayRef, ArrowPrimitiveType, BooleanArray, BooleanBufferBuilder,
+        BooleanBuilder, GenericBinaryArray, GenericBinaryBuilder, PrimitiveArray, PrimitiveBuilder,
+        StringArray, StringBuilder, UInt32Builder,
+    },
+    datatypes::{
+        Int16Type, Int32Type, Int64Type, Int8Type, Schema, UInt16Type, UInt32Type, UInt64Type,
+        UInt8Type,
     },
-    datatypes::{Int16Type, Int32Type, Int64Type, Int8Type, Schema},
 };
 
 use super::{column::Column, record::DynRecord, record_ref::DynRecordRef, Datatype};
@@ -35,6 +38,26 @@ impl ArrowArrays for DynRecordImmutableArrays {
         for field in schema.fields().iter().skip(2) {
             let datatype = Datatype::from(field.data_type());
             match datatype {
+                Datatype::UInt8 => {
+                    builders.push(Box::new(PrimitiveBuilder::<UInt8Type>::with_capacity(
+                        capacity,
+                    )));
+                }
+                Datatype::UInt16 => {
+                    builders.push(Box::new(PrimitiveBuilder::<UInt16Type>::with_capacity(
+                        capacity,
+                    )));
+                }
+                Datatype::UInt32 => {
+                    builders.push(Box::new(PrimitiveBuilder::<UInt32Type>::with_capacity(
+                        capacity,
+                    )));
+                }
+                Datatype::UInt64 => {
+                    builders.push(Box::new(PrimitiveBuilder::<UInt64Type>::with_capacity(
+                        capacity,
+                    )));
+                }
                 Datatype::Int8 => {
                     builders.push(Box::new(PrimitiveBuilder::<Int8Type>::with_capacity(
                         capacity,
@@ -98,34 +121,14 @@ impl ArrowArrays for DynRecordImmutableArrays {
                 let datatype = col.datatype;
                 let name = col.name.to_string();
                 let value: Arc<dyn Any> = match datatype {
-                    Datatype::Int8 => Arc::new(
-                        col.value
-                            .as_ref()
-                            .downcast_ref::<Int8Array>()
-                            .unwrap()
-                            .value(offset),
-                    ),
-                    Datatype::Int16 => Arc::new(
-                        col.value
-                            .as_ref()
-                            .downcast_ref::<Int16Array>()
-                            .unwrap()
-                            .value(offset),
-                    ),
-                    Datatype::Int32 => Arc::new(
-                        col.value
-                            .as_ref()
-                            .downcast_ref::<Int32Array>()
-                            .unwrap()
-                            .value(offset),
-                    ),
-                    Datatype::Int64 => Arc::new(
-                        col.value
-                            .as_ref()
-                            .downcast_ref::<Int64Array>()
-                            .unwrap()
-                            .value(offset),
-                    ),
+                    Datatype::UInt8 => Arc::new(Self::primitive_value::<UInt8Type>(col, offset)),
+                    Datatype::UInt16 => Arc::new(Self::primitive_value::<UInt16Type>(col, offset)),
+                    Datatype::UInt32 => Arc::new(Self::primitive_value::<UInt32Type>(col, offset)),
+                    Datatype::UInt64 => Arc::new(Self::primitive_value::<UInt64Type>(col, offset)),
+                    Datatype::Int8 => Arc::new(Self::primitive_value::<Int8Type>(col, offset)),
+                    Datatype::Int16 => Arc::new(Self::primitive_value::<Int16Type>(col, offset)),
+                    Datatype::Int32 => Arc::new(Self::primitive_value::<Int32Type>(col, offset)),
+                    Datatype::Int64 => Arc::new(Self::primitive_value::<Int64Type>(col, offset)),
                     Datatype::String => Arc::new(
                         col.value
                             .as_ref()
@@ -167,6 +170,18 @@ impl ArrowArrays for DynRecordImmutableArrays {
         &self.record_batch
     }
 }
+impl DynRecordImmutableArrays {
+    fn primitive_value<T>(col: &Column, offset: usize) -> T::Native
+    where
+        T: ArrowPrimitiveType,
+    {
+        col.value
+            .as_ref()
+            .downcast_ref::<PrimitiveArray<T>>()
+            .unwrap()
+            .value(offset)
+    }
+}
 
 pub struct DynRecordBuilder {
     builders: Vec<Box<dyn ArrayBuilder>>,
@@ -204,12 +219,40 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                     }
                     let datatype = col.datatype;
                     match datatype {
+                        Datatype::UInt8 => {
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<UInt8Type>>(builder);
+                            let value = col.value.as_ref().downcast_ref::<Option<u8>>().unwrap();
+                            match value {
+                                Some(value) => bd.append_value(*value),
+                                None => bd.append_null(),
+                            }
+                        }
+                        Datatype::UInt16 => {
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<UInt16Type>>(builder);
+                            let value = col.value.as_ref().downcast_ref::<Option<u16>>().unwrap();
+                            match value {
+                                Some(value) => bd.append_value(*value),
+                                None => bd.append_null(),
+                            }
+                        }
+                        Datatype::UInt32 => {
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<UInt32Type>>(builder);
+                            let value = col.value.as_ref().downcast_ref::<Option<u32>>().unwrap();
+                            match value {
+                                Some(value) => bd.append_value(*value),
+                                None => bd.append_null(),
+                            }
+                        }
+                        Datatype::UInt64 => {
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<UInt64Type>>(builder);
+                            let value = col.value.as_ref().downcast_ref::<Option<u64>>().unwrap();
+                            match value {
+                                Some(value) => bd.append_value(*value),
+                                None => bd.append_null(),
+                            }
+                        }
                         Datatype::Int8 => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int8Type>>()
-                                .unwrap();
-
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<Int8Type>>(builder);
                             let value = col.value.as_ref().downcast_ref::<Option<i8>>().unwrap();
                             match value {
                                 Some(value) => bd.append_value(*value),
@@ -217,10 +260,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::Int16 => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int16Type>>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<Int16Type>>(builder);
                             let value = col.value.as_ref().downcast_ref::<Option<i16>>().unwrap();
                             match value {
                                 Some(value) => bd.append_value(*value),
@@ -228,10 +268,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::Int32 => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int32Type>>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<Int32Type>>(builder);
                             let value = col.value.as_ref().downcast_ref::<Option<i32>>().unwrap();
                             match value {
                                 Some(value) => bd.append_value(*value),
@@ -239,10 +276,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::Int64 => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int64Type>>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<PrimitiveBuilder<Int64Type>>(builder);
                             let value = col.value.as_ref().downcast_ref::<Option<i64>>().unwrap();
                             match value {
                                 Some(value) => bd.append_value(*value),
@@ -250,10 +284,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::String => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<StringBuilder>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<StringBuilder>(builder);
                             let value =
                                 col.value.as_ref().downcast_ref::<Option<String>>().unwrap();
                             match value {
@@ -262,10 +293,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::Boolean => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<BooleanBuilder>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<BooleanBuilder>(builder);
                             let value = col.value.as_ref().downcast_ref::<Option<bool>>().unwrap();
                             match value {
                                 Some(value) => bd.append_value(*value),
@@ -273,10 +301,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                             }
                         }
                         Datatype::Bytes => {
-                            let bd = builder
-                                .as_any_mut()
-                                .downcast_mut::<GenericBinaryBuilder<i32>>()
-                                .unwrap();
+                            let bd = Self::as_builder_mut::<GenericBinaryBuilder<i32>>(builder);
                             let value = col
                                 .value
                                 .as_ref()
@@ -301,53 +326,48 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                         continue;
                     }
                     match datatype {
+                        Datatype::UInt8 => {
+                            Self::as_builder_mut::<PrimitiveBuilder<UInt8Type>>(builder)
+                                .append_value(u8::default());
+                        }
+                        Datatype::UInt16 => {
+                            Self::as_builder_mut::<PrimitiveBuilder<UInt16Type>>(builder)
+                                .append_value(u16::default());
+                        }
+                        Datatype::UInt32 => {
+                            Self::as_builder_mut::<PrimitiveBuilder<UInt32Type>>(builder)
+                                .append_value(u32::default());
+                        }
+                        Datatype::UInt64 => {
+                            Self::as_builder_mut::<PrimitiveBuilder<UInt64Type>>(builder)
+                                .append_value(u64::default());
+                        }
                         Datatype::Int8 => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int8Type>>()
-                                .unwrap()
+                            Self::as_builder_mut::<PrimitiveBuilder<Int8Type>>(builder)
                                 .append_value(i8::default());
                         }
                         Datatype::Int16 => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int16Type>>()
-                                .unwrap()
+                            Self::as_builder_mut::<PrimitiveBuilder<Int16Type>>(builder)
                                 .append_value(i16::default());
                         }
                         Datatype::Int32 => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int32Type>>()
-                                .unwrap()
+                            Self::as_builder_mut::<PrimitiveBuilder<Int32Type>>(builder)
                                 .append_value(i32::default());
                         }
                         Datatype::Int64 => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<PrimitiveBuilder<Int64Type>>()
-                                .unwrap()
+                            Self::as_builder_mut::<PrimitiveBuilder<Int64Type>>(builder)
                                 .append_value(i64::default());
                         }
                         Datatype::String => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<StringBuilder>()
-                                .unwrap()
+                            Self::as_builder_mut::<StringBuilder>(builder)
                                 .append_value(String::default());
                         }
                         Datatype::Boolean => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<BooleanBuilder>()
-                                .unwrap()
+                            Self::as_builder_mut::<BooleanBuilder>(builder)
                                 .append_value(bool::default());
                         }
                         Datatype::Bytes => {
-                            builder
-                                .as_any_mut()
-                                .downcast_mut::<GenericBinaryBuilder<i32>>()
-                                .unwrap()
+                            Self::as_builder_mut::<GenericBinaryBuilder<i32>>(builder)
                                 .append_value(Vec::<u8>::default());
                         }
                     }
@@ -363,54 +383,38 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
             .zip(self.datatypes.iter())
             .fold(size, |acc, (builder, datatype)| {
                 acc + match datatype {
+                    Datatype::UInt8 => mem::size_of_val(
+                        Self::as_builder::<PrimitiveBuilder<UInt8Type>>(builder).values_slice(),
+                    ),
+                    Datatype::UInt16 => mem::size_of_val(
+                        Self::as_builder::<PrimitiveBuilder<UInt16Type>>(builder).values_slice(),
+                    ),
+                    Datatype::UInt32 => mem::size_of_val(
+                        Self::as_builder::<PrimitiveBuilder<UInt32Type>>(builder).values_slice(),
+                    ),
+                    Datatype::UInt64 => mem::size_of_val(
+                        Self::as_builder::<PrimitiveBuilder<UInt64Type>>(builder).values_slice(),
+                    ),
                     Datatype::Int8 => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<PrimitiveBuilder<Int8Type>>()
-                            .unwrap()
-                            .values_slice(),
+                        Self::as_builder::<PrimitiveBuilder<Int8Type>>(builder).values_slice(),
                     ),
                     Datatype::Int16 => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<PrimitiveBuilder<Int16Type>>()
-                            .unwrap()
-                            .values_slice(),
+                        Self::as_builder::<PrimitiveBuilder<Int16Type>>(builder).values_slice(),
                     ),
                     Datatype::Int32 => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<PrimitiveBuilder<Int32Type>>()
-                            .unwrap()
-                            .values_slice(),
+                        Self::as_builder::<PrimitiveBuilder<Int32Type>>(builder).values_slice(),
                     ),
                     Datatype::Int64 => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<PrimitiveBuilder<Int64Type>>()
-                            .unwrap()
-                            .values_slice(),
-                    ),
-                    Datatype::String => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<StringBuilder>()
-                            .unwrap()
-                            .values_slice(),
-                    ),
-                    Datatype::Boolean => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<BooleanBuilder>()
-                            .unwrap()
-                            .values_slice(),
+                        Self::as_builder::<PrimitiveBuilder<Int64Type>>(builder).values_slice(),
                     ),
+                    Datatype::String => {
+                        mem::size_of_val(Self::as_builder::<StringBuilder>(builder).values_slice())
+                    }
+                    Datatype::Boolean => {
+                        mem::size_of_val(Self::as_builder::<BooleanBuilder>(builder).values_slice())
+                    }
                     Datatype::Bytes => mem::size_of_val(
-                        builder
-                            .as_any()
-                            .downcast_ref::<GenericBinaryBuilder<i32>>()
-                            .unwrap()
-                            .values_slice(),
+                        Self::as_builder::<GenericBinaryBuilder<i32>>(builder).values_slice(),
                     ),
                 }
             })
@@ -431,13 +435,57 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
             let field = self.schema.field(idx + 2);
             let is_nullable = field.is_nullable();
             match datatype {
+                Datatype::UInt8 => {
+                    let value = Arc::new(
+                        Self::as_builder_mut::<PrimitiveBuilder<UInt8Type>>(builder).finish(),
+                    );
+                    columns.push(Column {
+                        datatype: Datatype::UInt8,
+                        name: field.name().to_owned(),
+                        value: value.clone(),
+                        is_nullable,
+                    });
+                    array_refs.push(value);
+                }
+                Datatype::UInt16 => {
+                    let value = Arc::new(
+                        Self::as_builder_mut::<PrimitiveBuilder<UInt16Type>>(builder).finish(),
+                    );
+                    columns.push(Column {
+                        datatype: Datatype::UInt16,
+                        name: field.name().to_owned(),
+                        value: value.clone(),
+                        is_nullable,
+                    });
+                    array_refs.push(value);
+                }
+                Datatype::UInt32 => {
+                    let value = Arc::new(
+                        Self::as_builder_mut::<PrimitiveBuilder<UInt32Type>>(builder).finish(),
+                    );
+                    columns.push(Column {
+                        datatype: Datatype::UInt32,
+                        name: field.name().to_owned(),
+                        value: value.clone(),
+                        is_nullable,
+                    });
+                    array_refs.push(value);
+                }
+                Datatype::UInt64 => {
+                    let value = Arc::new(
+                        Self::as_builder_mut::<PrimitiveBuilder<UInt64Type>>(builder).finish(),
+                    );
+                    columns.push(Column {
+                        datatype: Datatype::UInt64,
+                        name: field.name().to_owned(),
+                        value: value.clone(),
+                        is_nullable,
+                    });
+                    array_refs.push(value);
+                }
                 Datatype::Int8 => {
                     let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<PrimitiveBuilder<Int8Type>>()
-                            .unwrap()
-                            .finish(),
+                        Self::as_builder_mut::<PrimitiveBuilder<Int8Type>>(builder).finish(),
                     );
                     columns.push(Column {
                         datatype: Datatype::Int8,
@@ -449,11 +497,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                 }
                 Datatype::Int16 => {
                     let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<PrimitiveBuilder<Int16Type>>()
-                            .unwrap()
-                            .finish(),
+                        Self::as_builder_mut::<PrimitiveBuilder<Int16Type>>(builder).finish(),
                     );
                     columns.push(Column {
                         datatype: Datatype::Int16,
@@ -465,11 +509,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                 }
                 Datatype::Int32 => {
                     let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<PrimitiveBuilder<Int32Type>>()
-                            .unwrap()
-                            .finish(),
+                        Self::as_builder_mut::<PrimitiveBuilder<Int32Type>>(builder).finish(),
                     );
                     columns.push(Column {
                         datatype: Datatype::Int32,
@@ -481,11 +521,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                 }
                 Datatype::Int64 => {
                     let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<PrimitiveBuilder<Int64Type>>()
-                            .unwrap()
-                            .finish(),
+                        Self::as_builder_mut::<PrimitiveBuilder<Int64Type>>(builder).finish(),
                     );
                     columns.push(Column {
                         datatype: Datatype::Int64,
@@ -496,13 +532,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                     array_refs.push(value);
                 }
                 Datatype::String => {
-                    let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<StringBuilder>()
-                            .unwrap()
-                            .finish(),
-                    );
+                    let value = Arc::new(Self::as_builder_mut::<StringBuilder>(builder).finish());
                     columns.push(Column {
                         datatype: Datatype::String,
                         name: field.name().to_owned(),
@@ -512,13 +542,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                     array_refs.push(value);
                 }
                 Datatype::Boolean => {
-                    let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<BooleanBuilder>()
-                            .unwrap()
-                            .finish(),
-                    );
+                    let value = Arc::new(Self::as_builder_mut::<BooleanBuilder>(builder).finish());
                     columns.push(Column {
                         datatype: Datatype::Boolean,
                         name: field.name().to_owned(),
@@ -529,11 +553,7 @@ impl Builder<DynRecordImmutableArrays> for DynRecordBuilder {
                 }
                 Datatype::Bytes => {
                     let value = Arc::new(
-                        builder
-                            .as_any_mut()
-                            .downcast_mut::<GenericBinaryBuilder<i32>>()
-                            .unwrap()
-                            .finish(),
+                        Self::as_builder_mut::<GenericBinaryBuilder<i32>>(builder).finish(),
                     );
                     columns.push(Column {
                         datatype: Datatype::Bytes,
@@ -574,43 +594,44 @@ impl DynRecordBuilder {
         let datatype = self.datatypes.get_mut(primary_key_index).unwrap();
         let col = key.value;
         match datatype {
-            Datatype::Int8 => builder
-                .as_any_mut()
-                .downcast_mut::<PrimitiveBuilder<Int8Type>>()
-                .unwrap()
+            Datatype::UInt8 => Self::as_builder_mut::<PrimitiveBuilder<UInt8Type>>(builder)
+                .append_value(*col.value.as_ref().downcast_ref::<u8>().unwrap()),
+            Datatype::UInt16 => Self::as_builder_mut::<PrimitiveBuilder<UInt16Type>>(builder)
+                .append_value(*col.value.as_ref().downcast_ref::<u16>().unwrap()),
+            Datatype::UInt32 => Self::as_builder_mut::<PrimitiveBuilder<UInt32Type>>(builder)
+                .append_value(*col.value.as_ref().downcast_ref::<u32>().unwrap()),
+            Datatype::UInt64 => Self::as_builder_mut::<PrimitiveBuilder<UInt64Type>>(builder)
+                .append_value(*col.value.as_ref().downcast_ref::<u64>().unwrap()),
+            Datatype::Int8 => Self::as_builder_mut::<PrimitiveBuilder<Int8Type>>(builder)
                 .append_value(*col.value.as_ref().downcast_ref::<i8>().unwrap()),
-            Datatype::Int16 => builder
-                .as_any_mut()
-                .downcast_mut::<PrimitiveBuilder<Int16Type>>()
-                .unwrap()
+            Datatype::Int16 => Self::as_builder_mut::<PrimitiveBuilder<Int16Type>>(builder)
                 .append_value(*col.value.as_ref().downcast_ref::<i16>().unwrap()),
-            Datatype::Int32 => builder
-                .as_any_mut()
-                .downcast_mut::<PrimitiveBuilder<Int32Type>>()
-                .unwrap()
+            Datatype::Int32 => Self::as_builder_mut::<PrimitiveBuilder<Int32Type>>(builder)
                 .append_value(*col.value.as_ref().downcast_ref::<i32>().unwrap()),
-            Datatype::Int64 => builder
-                .as_any_mut()
-                .downcast_mut::<PrimitiveBuilder<Int64Type>>()
-                .unwrap()
+            Datatype::Int64 => Self::as_builder_mut::<PrimitiveBuilder<Int64Type>>(builder)
                 .append_value(*col.value.as_ref().downcast_ref::<i64>().unwrap()),
-            Datatype::String => builder
-                .as_any_mut()
-                .downcast_mut::<StringBuilder>()
-                .unwrap()
+            Datatype::String => Self::as_builder_mut::<StringBuilder>(builder)
                 .append_value(col.value.as_ref().downcast_ref::<String>().unwrap()),
-            Datatype::Boolean => builder
-                .as_any_mut()
-                .downcast_mut::<BooleanBuilder>()
-                .unwrap()
+            Datatype::Boolean => Self::as_builder_mut::<BooleanBuilder>(builder)
                 .append_value(*col.value.as_ref().downcast_ref::<bool>().unwrap()),
-            Datatype::Bytes => builder
-                .as_any_mut()
-                .downcast_mut::<GenericBinaryBuilder<i32>>()
-                .unwrap()
+            Datatype::Bytes => Self::as_builder_mut::<GenericBinaryBuilder<i32>>(builder)
                 .append_value(col.value.as_ref().downcast_ref::<Vec<u8>>().unwrap()),
         };
     }
+
+    fn as_builder<T>(builder: &dyn ArrayBuilder) -> &T
+    where
+        T: ArrayBuilder,
+    {
+        builder.as_any().downcast_ref::<T>().unwrap()
+    }
+
+    fn as_builder_mut<T>(builder: &mut dyn ArrayBuilder) -> &mut T
+    where
+        T: ArrayBuilder,
+    {
+        builder.as_any_mut().downcast_mut::<T>().unwrap()
+    }
 }
 
 unsafe impl Send for DynRecordBuilder {}
diff --git a/src/record/runtime/column.rs b/src/record/runtime/column.rs
index ab47111e..6db8bc4a 100644
--- a/src/record/runtime/column.rs
+++ b/src/record/runtime/column.rs
@@ -3,7 +3,7 @@ use std::{any::Any, fmt::Debug, hash::Hash, sync::Arc};
 use arrow::{
     array::{
         BooleanArray, GenericBinaryArray, Int16Array, Int32Array, Int64Array, Int8Array,
-        StringArray,
+        StringArray, UInt16Array, UInt32Array, UInt64Array, UInt8Array,
     },
     datatypes::{DataType, Field},
 };
@@ -53,8 +53,18 @@ impl Column {
         }
     }
 
-    pub fn with_none_value(datatype: Datatype, name: String, is_nullable: bool) -> Self {
+    pub(crate) fn with_none_value(datatype: Datatype, name: String, is_nullable: bool) -> Self {
         match datatype {
+            Datatype::UInt8 => Self::new(datatype, name, Arc::<Option<u8>>::new(None), is_nullable),
+            Datatype::UInt16 => {
+                Self::new(datatype, name, Arc::<Option<u16>>::new(None), is_nullable)
+            }
+            Datatype::UInt32 => {
+                Self::new(datatype, name, Arc::<Option<u32>>::new(None), is_nullable)
+            }
+            Datatype::UInt64 => {
+                Self::new(datatype, name, Arc::<Option<u64>>::new(None), is_nullable)
+            }
             Datatype::Int8 => Self::new(datatype, name, Arc::<Option<i8>>::new(None), is_nullable),
             Datatype::Int16 => {
                 Self::new(datatype, name, Arc::<Option<i16>>::new(None), is_nullable)
@@ -312,25 +322,33 @@ macro_rules! implement_encode_col {
 impl Column {
     fn tag(datatype: Datatype) -> u8 {
         match datatype {
-            Datatype::Int8 => 0,
-            Datatype::Int16 => 1,
-            Datatype::Int32 => 2,
-            Datatype::Int64 => 3,
-            Datatype::String => 4,
-            Datatype::Boolean => 5,
-            Datatype::Bytes => 6,
+            Datatype::UInt8 => 0,
+            Datatype::UInt16 => 1,
+            Datatype::UInt32 => 2,
+            Datatype::UInt64 => 3,
+            Datatype::Int8 => 4,
+            Datatype::Int16 => 5,
+            Datatype::Int32 => 6,
+            Datatype::Int64 => 7,
+            Datatype::String => 8,
+            Datatype::Boolean => 9,
+            Datatype::Bytes => 10,
         }
     }
 
     fn tag_to_datatype(tag: u8) -> Datatype {
         match tag {
-            0 => Datatype::Int8,
-            1 => Datatype::Int16,
-            2 => Datatype::Int32,
-            3 => Datatype::Int64,
-            4 => Datatype::String,
-            5 => Datatype::Boolean,
-            6 => Datatype::Bytes,
+            0 => Datatype::UInt8,
+            1 => Datatype::UInt16,
+            2 => Datatype::UInt32,
+            3 => Datatype::UInt64,
+            4 => Datatype::Int8,
+            5 => Datatype::Int16,
+            6 => Datatype::Int32,
+            7 => Datatype::Int64,
+            8 => Datatype::String,
+            9 => Datatype::Boolean,
+            10 => Datatype::Bytes,
             _ => panic!("invalid datatype tag"),
         }
     }
@@ -339,6 +357,10 @@ impl Column {
 impl From<&Column> for Field {
     fn from(col: &Column) -> Self {
         match col.datatype {
+            Datatype::UInt8 => Field::new(&col.name, DataType::UInt8, col.is_nullable),
+            Datatype::UInt16 => Field::new(&col.name, DataType::UInt16, col.is_nullable),
+            Datatype::UInt32 => Field::new(&col.name, DataType::UInt32, col.is_nullable),
+            Datatype::UInt64 => Field::new(&col.name, DataType::UInt64, col.is_nullable),
             Datatype::Int8 => Field::new(&col.name, DataType::Int8, col.is_nullable),
             Datatype::Int16 => Field::new(&col.name, DataType::Int16, col.is_nullable),
             Datatype::Int32 => Field::new(&col.name, DataType::Int32, col.is_nullable),
@@ -355,6 +377,10 @@ macro_rules! for_datatype {
     ($macro:tt $(, $x:tt)*) => {
         $macro! {
             [$($x),*],
+                { u8, UInt8 },
+                { u16, UInt16 },
+                { u32, UInt32 },
+                { u64, UInt64 },
                 { i8, Int8 },
                 { i16, Int16 },
                 { i32, Int32 },
@@ -367,6 +393,7 @@ macro_rules! for_datatype {
 }
 
 implement_key_col!(
+    { u8, UInt8, UInt8Array }, { u16, UInt16, UInt16Array }, { u32, UInt32, UInt32Array }, { u64, UInt64, UInt64Array },
     { i8, Int8, Int8Array }, { i16, Int16, Int16Array }, { i32, Int32, Int32Array }, { i64, Int64, Int64Array }
 );
 for_datatype! { implement_col }
diff --git a/src/record/runtime/mod.rs b/src/record/runtime/mod.rs
index a1f780e2..90e8b304 100644
--- a/src/record/runtime/mod.rs
+++ b/src/record/runtime/mod.rs
@@ -10,6 +10,10 @@ pub use record_ref::*;
 
 #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Datatype {
+    UInt8,
+    UInt16,
+    UInt32,
+    UInt64,
     Int8,
     Int16,
     Int32,
@@ -22,6 +26,10 @@ pub enum Datatype {
 impl From<&DataType> for Datatype {
     fn from(datatype: &DataType) -> Self {
         match datatype {
+            DataType::UInt8 => Datatype::UInt8,
+            DataType::UInt16 => Datatype::UInt16,
+            DataType::UInt32 => Datatype::UInt32,
+            DataType::UInt64 => Datatype::UInt64,
             DataType::Int8 => Datatype::Int8,
             DataType::Int16 => Datatype::Int16,
             DataType::Int32 => Datatype::Int32,
diff --git a/src/record/runtime/record.rs b/src/record/runtime/record.rs
index 447af1f7..b2f67b6d 100644
--- a/src/record/runtime/record.rs
+++ b/src/record/runtime/record.rs
@@ -56,6 +56,22 @@ impl DynRecord {
         let mut columns = vec![];
         for desc in column_descs.iter() {
             let value: Arc<dyn Any> = match desc.datatype {
+                Datatype::UInt8 => match desc.is_nullable {
+                    true => Arc::<Option<u8>>::new(None),
+                    false => Arc::new(u8::default()),
+                },
+                Datatype::UInt16 => match desc.is_nullable {
+                    true => Arc::<Option<u16>>::new(None),
+                    false => Arc::new(u16::default()),
+                },
+                Datatype::UInt32 => match desc.is_nullable {
+                    true => Arc::<Option<u32>>::new(None),
+                    false => Arc::new(u32::default()),
+                },
+                Datatype::UInt64 => match desc.is_nullable {
+                    true => Arc::<Option<u64>>::new(None),
+                    false => Arc::new(u64::default()),
+                },
                 Datatype::Int8 => match desc.is_nullable {
                     true => Arc::<Option<i8>>::new(None),
                     false => Arc::new(i8::default()),
@@ -112,6 +128,22 @@ impl Decode for DynRecord {
             let mut col = Column::decode(reader).await?;
             if i != primary_index && !col.is_nullable {
                 match col.datatype {
+                    Datatype::UInt8 => {
+                        let value = col.value.as_ref().downcast_ref::<Option<u8>>().unwrap();
+                        col.value = Arc::new(value.unwrap());
+                    }
+                    Datatype::UInt16 => {
+                        let value = col.value.as_ref().downcast_ref::<Option<u16>>().unwrap();
+                        col.value = Arc::new(value.unwrap());
+                    }
+                    Datatype::UInt32 => {
+                        let value = col.value.as_ref().downcast_ref::<Option<u32>>().unwrap();
+                        col.value = Arc::new(value.unwrap());
+                    }
+                    Datatype::UInt64 => {
+                        let value = col.value.as_ref().downcast_ref::<Option<u64>>().unwrap();
+                        col.value = Arc::new(value.unwrap());
+                    }
                     Datatype::Int8 => {
                         let value = col.value.as_ref().downcast_ref::<Option<i8>>().unwrap();
                         col.value = Arc::new(value.unwrap());
@@ -179,6 +211,18 @@ impl Record for DynRecord {
             let mut value = col.value.clone();
             if idx != self.primary_index && !is_nullable {
                 value = match datatype {
+                    Datatype::UInt8 => {
+                        Arc::new(Some(*col.value.as_ref().downcast_ref::<u8>().unwrap()))
+                    }
+                    Datatype::UInt16 => {
+                        Arc::new(Some(*col.value.as_ref().downcast_ref::<u16>().unwrap()))
+                    }
+                    Datatype::UInt32 => {
+                        Arc::new(Some(*col.value.as_ref().downcast_ref::<u32>().unwrap()))
+                    }
+                    Datatype::UInt64 => {
+                        Arc::new(Some(*col.value.as_ref().downcast_ref::<u64>().unwrap()))
+                    }
                     Datatype::Int8 => {
                         Arc::new(Some(*col.value.as_ref().downcast_ref::<i8>().unwrap()))
                     }
diff --git a/src/record/runtime/record_ref.rs b/src/record/runtime/record_ref.rs
index f0bc81b7..bd64d6f8 100644
--- a/src/record/runtime/record_ref.rs
+++ b/src/record/runtime/record_ref.rs
@@ -1,8 +1,11 @@
 use std::{any::Any, marker::PhantomData, mem, sync::Arc};
 
 use arrow::{
-    array::{Array, AsArray},
-    datatypes::Schema,
+    array::{Array, ArrayRef, ArrowPrimitiveType, AsArray},
+    datatypes::{
+        Int16Type, Int32Type, Int64Type, Int8Type, Schema, UInt16Type, UInt32Type, UInt64Type,
+        UInt8Type,
+    },
 };
 use fusio::Write;
 
@@ -105,50 +108,62 @@ impl<'r> RecordRef<'r> for DynRecordRef<'r> {
             let col = record_batch.column(batch_field.unwrap().0);
             let is_nullable = field.is_nullable();
             let value = match datatype {
-                Datatype::Int8 => {
-                    let v = col.as_primitive::<arrow::datatypes::Int8Type>();
-
-                    if primary_index == idx - 2 {
-                        Arc::new(v.value(offset)) as Arc<dyn Any>
-                    } else {
-                        let value = (!v.is_null(offset) && projection_mask.leaf_included(idx))
-                            .then_some(v.value(offset));
-                        Arc::new(value) as Arc<dyn Any>
-                    }
-                }
-                Datatype::Int16 => {
-                    let v = col.as_primitive::<arrow::datatypes::Int16Type>();
-
-                    if primary_index == idx - 2 {
-                        Arc::new(v.value(offset)) as Arc<dyn Any>
-                    } else {
-                        let value = (!v.is_null(offset) && projection_mask.leaf_included(idx))
-                            .then_some(v.value(offset));
-                        Arc::new(value) as Arc<dyn Any>
-                    }
-                }
-                Datatype::Int32 => {
-                    let v = col.as_primitive::<arrow::datatypes::Int32Type>();
-
-                    if primary_index == idx - 2 {
-                        Arc::new(v.value(offset)) as Arc<dyn Any>
-                    } else {
-                        let value = (!v.is_null(offset) && projection_mask.leaf_included(idx))
-                            .then_some(v.value(offset));
-                        Arc::new(value) as Arc<dyn Any>
-                    }
-                }
-                Datatype::Int64 => {
-                    let v = col.as_primitive::<arrow::datatypes::Int64Type>();
-
-                    if primary_index == idx - 2 {
-                        Arc::new(v.value(offset)) as Arc<dyn Any>
-                    } else {
-                        let value = (!v.is_null(offset) && projection_mask.leaf_included(idx))
-                            .then_some(v.value(offset));
-                        Arc::new(value) as Arc<dyn Any>
-                    }
-                }
+                Datatype::UInt8 => Self::primitive_value::<UInt8Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::UInt16 => Self::primitive_value::<UInt16Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::UInt32 => Self::primitive_value::<UInt32Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::UInt64 => Self::primitive_value::<UInt64Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::Int8 => Self::primitive_value::<Int8Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::Int16 => Self::primitive_value::<Int16Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::Int32 => Self::primitive_value::<Int32Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
+                Datatype::Int64 => Self::primitive_value::<Int64Type>(
+                    col,
+                    offset,
+                    idx,
+                    projection_mask,
+                    primary_index == idx - 2,
+                ),
                 Datatype::String => {
                     let v = col.as_string::<i32>();
 
@@ -202,6 +217,10 @@ impl<'r> RecordRef<'r> for DynRecordRef<'r> {
         for (idx, col) in self.columns.iter_mut().enumerate() {
             if idx != self.primary_index && !projection_mask.leaf_included(idx + 2) {
                 match col.datatype {
+                    Datatype::UInt8 => col.value = Arc::<Option<u8>>::new(None),
+                    Datatype::UInt16 => col.value = Arc::<Option<u16>>::new(None),
+                    Datatype::UInt32 => col.value = Arc::<Option<u32>>::new(None),
+                    Datatype::UInt64 => col.value = Arc::<Option<u64>>::new(None),
                     Datatype::Int8 => col.value = Arc::<Option<i8>>::new(None),
                     Datatype::Int16 => col.value = Arc::<Option<i16>>::new(None),
                     Datatype::Int32 => col.value = Arc::<Option<i32>>::new(None),
@@ -215,5 +234,28 @@ impl<'r> RecordRef<'r> for DynRecordRef<'r> {
     }
 }
 
+impl<'r> DynRecordRef<'r> {
+    fn primitive_value<T>(
+        col: &ArrayRef,
+        offset: usize,
+        idx: usize,
+        projection_mask: &'r parquet::arrow::ProjectionMask,
+        primary: bool,
+    ) -> Arc<dyn Any>
+    where
+        T: ArrowPrimitiveType,
+    {
+        let v = col.as_primitive::<T>();
+
+        if primary {
+            Arc::new(v.value(offset)) as Arc<dyn Any>
+        } else {
+            let value = (!v.is_null(offset) && projection_mask.leaf_included(idx))
+                .then_some(v.value(offset));
+            Arc::new(value) as Arc<dyn Any>
+        }
+    }
+}
+
 unsafe impl<'r> Send for DynRecordRef<'r> {}
 unsafe impl<'r> Sync for DynRecordRef<'r> {}


Test patch:
diff --git a/bindings/python/tests/test_db.py b/bindings/python/tests/test_db.py
new file mode 100644
index 00000000..c83d8360
--- /dev/null
+++ b/bindings/python/tests/test_db.py
@@ -0,0 +1,64 @@
+import pytest
+import tempfile
+from tonbo import DbOption, Column, DataType, Record, RecordBatch, TonboDB
+
+
+@Record
+class User:
+    age = Column(DataType.Int8, name="age", primary_key=True)
+    height = Column(DataType.Int16, name="height", nullable=True)
+    weight = Column(DataType.Int32, name="weight", nullable=False)
+
+
+def build_db():
+    temp_dir = tempfile.TemporaryDirectory()
+    return TonboDB(DbOption(temp_dir.name), User())
+
+
+@pytest.mark.asyncio
+async def test_db_write():
+    db = build_db()
+    for i in range(0, 100):
+        await db.insert(User(age=i, height=i * 10, weight=i * 20))
+
+
+@pytest.mark.asyncio
+async def test_db_read():
+    db = build_db()
+    for i in range(0, 100):
+        await db.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    for i in range(0, 100):
+        user = await db.get(i)
+        assert user == {"age": i, "height": i * 10, "weight": i * 20}
+
+
+@pytest.mark.asyncio
+async def test_db_write_batch():
+    db = build_db()
+    batch = RecordBatch()
+    for i in range(0, 100):
+        batch.append(User(age=i, height=i * 10, weight=i * 20))
+    await db.insert_batch(batch)
+
+    for i in range(0, 100):
+        user = await db.get(i)
+        assert user == {"age": i, "height": i * 10, "weight": i * 20}
+
+
+@pytest.mark.asyncio
+async def test_db_remove():
+    db = build_db()
+    for i in range(0, 100):
+        if i % 2 == 0:
+            await db.insert(User(age=i, height=i * 10, weight=i * 20))
+        # else:
+        #     await db.remove(i)
+
+    for i in range(0, 100):
+        print(i)
+        user = await db.get(i)
+        if i % 2 == 0:
+            assert user == {"age": i, "height": i * 10, "weight": i * 20}
+        else:
+            assert user is None
diff --git a/bindings/python/tests/test_flush.py b/bindings/python/tests/test_flush.py
new file mode 100644
index 00000000..f65febc1
--- /dev/null
+++ b/bindings/python/tests/test_flush.py
@@ -0,0 +1,46 @@
+import pytest
+import tempfile
+from tonbo import DbOption, Column, DataType, Record, TonboDB
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    name = Column(DataType.String, name="name")
+    email = Column(DataType.String, name="email", nullable=True)
+    age = Column(DataType.Int8, name="age")
+    data = Column(DataType.Bytes, name="data")
+
+
+@pytest.mark.asyncio
+async def test_flush():
+    temp_dir = tempfile.TemporaryDirectory()
+    option = DbOption(temp_dir.name)
+    option.immutable_chunk_num = 1
+    option.major_threshold_with_sst_size = 3
+    option.level_sst_magnification = 1
+    option.max_sst_file_size = 1 * 1024
+    db = TonboDB(option, User())
+    # db = build_db()
+    for i in range(0, 1000):
+        if i % 100 == 0:
+            await db.flush()
+        await db.insert(
+            User(
+                id=i,
+                age=i % 128,
+                name=str(i * 10),
+                email=str(i * 20),
+                data=b"Hello Tonbo!",
+            )
+        )
+
+    for i in range(0, 1000):
+        user = await db.get(i)
+        assert user == {
+            "id": i,
+            "name": str(i * 10),
+            "email": str(i * 20),
+            "age": i % 128,
+            "data": b"Hello Tonbo!",
+        }
diff --git a/bindings/python/tests/test_table_level.py b/bindings/python/tests/test_table_level.py
new file mode 100644
index 00000000..6f4a4d34
--- /dev/null
+++ b/bindings/python/tests/test_table_level.py
@@ -0,0 +1,69 @@
+import tempfile
+import pytest
+from tonbo import TonboDB, DbOption, Column, DataType, Record
+from tonbo.fs import from_filesystem_path, parse, from_url_path, FsOptions
+from tonbo.error import PathParseError
+
+
+@Record
+class User:
+    id = Column(DataType.Int64, name="id", primary_key=True)
+    name = Column(DataType.String, name="name")
+    email = Column(DataType.String, name="email", nullable=True)
+    age = Column(DataType.Int8, name="age")
+    data = Column(DataType.Bytes, name="data")
+
+
+def test_parse():
+    assert parse("/foo/bar") == "foo/bar"
+    with pytest.raises(PathParseError):
+        parse("//foo/bar")
+
+
+def test_from_url_path():
+    assert from_url_path("foo%20bar") == "foo bar"
+    assert from_url_path("foo%2F%252E%252E%2Fbar") == "foo/%2E%2E/bar"
+    assert from_url_path("foo/%252E%252E/bar") == "foo/%2E%2E/bar"
+    assert from_url_path("%48%45%4C%4C%4F") == "HELLO"
+    with pytest.raises(PathParseError):
+        from_url_path("foo/%2E%2E/bar")
+
+
+@pytest.mark.asyncio
+async def test_table_level_local():
+    temp_dir = tempfile.TemporaryDirectory()
+    temp_dir0 = tempfile.TemporaryDirectory()
+    temp_dir1 = tempfile.TemporaryDirectory()
+
+    option = DbOption(from_filesystem_path(temp_dir.name))
+    option.level_path(0, from_filesystem_path(temp_dir0.name), FsOptions.Local())
+    option.level_path(1, from_filesystem_path(temp_dir1.name), FsOptions.Local())
+
+    option.immutable_chunk_num = 1
+    option.major_threshold_with_sst_size = 3
+    option.level_sst_magnification = 1
+    option.max_sst_file_size = 1 * 1024
+
+    db = TonboDB(option, User())
+    for i in range(0, 1000):
+        if i % 50 == 0:
+            await db.flush()
+        await db.insert(
+            User(
+                id=i,
+                age=i % 128,
+                name=str(i * 10),
+                email=str(i * 20),
+                data=b"Hello Tonbo!",
+            )
+        )
+
+    for i in range(0, 1000):
+        user = await db.get(i)
+        assert user == {
+            "id": i,
+            "name": str(i * 10),
+            "email": str(i * 20),
+            "age": i % 128,
+            "data": b"Hello Tonbo!",
+        }
diff --git a/bindings/python/tests/test_txn.py b/bindings/python/tests/test_txn.py
new file mode 100644
index 00000000..c2643664
--- /dev/null
+++ b/bindings/python/tests/test_txn.py
@@ -0,0 +1,150 @@
+import pytest
+import tempfile
+from tonbo import DbOption, Column, DataType, Record, TonboDB, Bound
+from tonbo.error import RepeatedCommitError, WriteConflictError
+
+
+@Record
+class User:
+    age = Column(DataType.Int8, name="age", primary_key=True)
+    height = Column(DataType.Int16, name="height", nullable=True)
+    weight = Column(DataType.Int32, name="weight", nullable=False)
+
+
+def build_db():
+    temp_dir = tempfile.TemporaryDirectory()
+    return TonboDB(DbOption(temp_dir.name), User())
+
+
+@pytest.mark.asyncio
+async def test_txn_write():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+    await txn.commit()
+
+
+@pytest.mark.asyncio
+async def test_txn_read_write():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 11, weight=i * 22))
+
+    for i in range(0, 100):
+        user = await txn.get(i)
+        assert user == {"age": i, "height": i * 11, "weight": i * 22}
+
+
+@pytest.mark.asyncio
+async def test_txn_remove():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        if i % 2 == 0:
+            txn.insert(User(age=i, height=i * 10, weight=i * 20))
+        else:
+            txn.remove(i)
+
+    for i in range(0, 100):
+        user = await txn.get(i)
+        if i % 2 == 0:
+            assert user == {"age": i, "height": i * 10, "weight": i * 20}
+        else:
+            assert user is None
+
+
+@pytest.mark.asyncio
+async def test_txn_scan():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    scan = await txn.scan(Bound.Included(10), Bound.Excluded(75))
+    i = 10
+    async for user in scan:
+        assert user == {"age": i, "height": i * 10, "weight": i * 20}
+        i += 1
+    assert i == 75
+
+
+@pytest.mark.asyncio
+async def test_txn_projection():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    scan = await txn.scan(None, Bound.Excluded(75), projection=["age", "height"])
+    i = 0
+    async for user in scan:
+        assert user == {"age": i, "height": i * 10, "weight": None}
+        i += 1
+
+
+@pytest.mark.asyncio
+async def test_txn_limit():
+    db = build_db()
+    txn = await db.transaction()
+    for i in range(0, 100):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    scan = await txn.scan(None, Bound.Excluded(75), limit=10)
+    i = 0
+    async for user in scan:
+        assert user == {"age": i, "height": i * 10, "weight": i * 20}
+        i += 1
+    assert i == 10
+
+
+@pytest.mark.asyncio
+async def test_repeated_commit():
+    db = build_db()
+    txn = await db.transaction()
+    await txn.commit()
+
+    with pytest.raises(RepeatedCommitError):
+        await txn.commit()
+
+
+@pytest.mark.asyncio
+async def test_txn_write_conflict():
+    db = build_db()
+    txn = await db.transaction()
+    txn2 = await db.transaction()
+    for i in range(0, 10):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    for i in range(0, 10):
+        txn2.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    await txn2.commit()
+    with pytest.raises(WriteConflictError):
+        await txn.commit()
+
+
+@pytest.mark.asyncio
+async def test_txn_cannot_read_future():
+    db = build_db()
+    txn = await db.transaction()
+    txn2 = await db.transaction()
+    for i in range(0, 10):
+        txn.insert(User(age=i, height=i * 10, weight=i * 20))
+
+    for i in range(0, 10):
+        txn2.insert(User(age=i, height=i * 20, weight=i * 40))
+
+    await txn2.commit()
+    for i in range(0, 10):
+        user = await txn.get(i)
+        # can not read data in the future
+        assert user == { "age": i, "height": i * 10, "weight": i * 20 }
+
+    txn3 = await db.transaction()
+    for i in range(0, 10):
+        user = await txn3.get(i)
+        assert user == { "age": i, "height": i * 20, "weight": i * 40 }