🦔

TiDB for AI : pytidbで既存テーブルを扱う方法

2025/08/05に公開

TiDB

pytidb

tech

いままで３回にわたってAI に特化したTiDB用 Python SDKであるpytidbを触ってきました。

https://pingcap.github.io/ai/
こちらに存在しているサンプルはどれも以下の形式でtableオブジェクトをセットしています。

class Item(TableModel):
    __tablename__ = "items"
    id: int = Field(primary_key=True)
    title: str = FullTextField(fts_parser="MULTILINGUAL")

table = client.create_table(schema=Item, mode="overwrite")

既存テーブルをオブジェクトにセットする方法がわからず調べていたところpytidbのGitHubレポジトリを中の関口さんに教えていただきました。

見ているとopen_tableオペレーションが存在していました。名前からしてこれが既存テーブルをオブジェクトとして返すオペレーションのようなのでやってみます。

さっそくやってみる

open_table

connect.py

from pytidb import TiDBClient
from pytidb.embeddings import EmbeddingFunction
from pytidb.schema import TableModel, Field, VectorField
from pytidb.schema import FullTextField
import json

# データベース接続
client = TiDBClient.connect(
    host="gateway01.eu-central-1.prod.aws.tidbcloud.com",
    port=4000,
    username="3CLxN311CtrMaPB.root",
    password="L5hExxxxx",
    database="test",
)

class Item(TableModel):
    __tablename__ = "items"
    id: int = Field(primary_key=True)
    title: str = FullTextField(fts_parser="MULTILINGUAL")

# 既存のテーブルを開く（テーブル名のみ）
table = client.open_table("items")

print(f"テーブルオブジェクト: {table}")

results = table.search("Bluetooth Headphones", search_type="fulltext").limit(3).to_list()
print(json.dumps(results, indent=2, ensure_ascii=False))

 python connect.py
テーブルオブジェクト: <pytidb.table.Table object at 0x793573057fe0>
[
  {
    "id": 2,
    "title": "Bluetooth 5.3 Headphones, Noise Cancelling, Immersive sound, Comfortable",
    "_match_score": 3.7390857,
    "_score": 3.7390857
  },
  {
    "id": 5,
    "title": "Wired Headphones, Studio-grade, HD sound, Comfortable, Pro music experience",
    "_match_score": 1.9798478,
    "_score": 1.9798478
  },
  {
    "id": 1,
    "title": "Bluetooth Earphones, HiFi sound, 48h battery, Fast charge, Low latency",
    "_match_score": 1.620981,
    "_score": 1.620981
  }
]

無事動作しました！

has_table

それだけでは短いので似たようなhas_tableも試します。
Gitでは以下の様になっておりこれはテーブルが存在するかどうかのbooleanが戻るようです。

    def has_table(self, table_name: str) -> bool:
        return self._inspector.has_table(table_name)

先ほどのconnect.pyを以下に置き換えます。

connect.py

from pytidb import TiDBClient
from pytidb.embeddings import EmbeddingFunction
from pytidb.schema import TableModel, Field, VectorField
from pytidb.schema import FullTextField
import json

# データベース接続
client = TiDBClient.connect(
   host="gateway01.eu-central-1.prod.aws.tidbcloud.com",
   port=4000,
   username="3CLxN311CtrMaPB.root",
   password="L5hExxxxx",
   database="test",
)

# テーブルの存在確認
table_exists = client.has_table("items")
print(f"テーブル存在確認: {table_exists}")

# 既存のテーブルを開く
if table_exists:
   table = client.open_table("items")

list_tables

テーブル一覧を出力するオペレーションも試しておきます。

connect.py

from pytidb import TiDBClient
from pytidb.embeddings import EmbeddingFunction
from pytidb.schema import TableModel, Field, VectorField
from pytidb.schema import FullTextField
import json

# データベース接続
client = TiDBClient.connect(
    host="gateway01.eu-central-1.prod.aws.tidbcloud.com",
    port=4000,
    username="3CLxN311CtrMaPB.root",
    password="L5hExxxxx",
    database="test",
)

# 全テーブル一覧を取得
all_tables = client.list_tables()
print(f"データベース内のテーブル一覧: {all_tables}")

# テーブルの存在確認
table_exists = client.has_table("items")
print(f"テーブル存在確認: {table_exists}")

# 既存のテーブルを開く
if table_exists:
    table = client.open_table("items")

python connect.py
データベース内のテーブル一覧: ['items']
テーブル存在確認: True

さっそくやってみる

open_table

has_table

list_tables

Discussion