pyobvector Python SDKインターフェースの説明|V4.3.5|OceanBaseデータベース|OceanBaseデータベースドキュメント|分散データベース

pyobvector Python SDKインターフェースの説明

最終更新日：2026-04-09 02:53:56 更新

pyobvectorはOceanBaseベクトルストレージ機能のpython SDKであり、2つの使用モードを提供します。

pymilvus互換モード：MilvusLikeClientオブジェクトを使用してデータベースを操作し、軽量なMilvusClientと互換性のある一般的なインターフェースを提供します。
SQLAlchemy拡張モード：ObVecClientオブジェクトを使用してデータベースを操作し、リレーショナルデータベース用のpython SDK拡張機能を提供します。

本記事では、2つのモードにおける使用インターフェースと例をそれぞれ紹介します。

MilvusLikeClient

コンストラクター


def __init__(
    self,
    uri: str = "127.0.0.1:2881",
    user: str = "root@test",
    password: str = "",
    db_name: str = "test",
    **kwargs,
)

collection関連インターフェース

API	パラメータ記述	例
`def create_schema(self, **kwargs) -> CollectionSchema:`	CollectionSchemaオブジェクトを作成しますパラメータなしで、空のモード定義を初期化できます。選択可能なパラメータについては以下のとおりです： fields：FieldSchemaのリスト(詳細は以下のadd_schemaインターフェースを参照してください) partitions：パーティションルール(詳細についてはObPartitionを使用したパーティションルールの定義に関する章を参照してください) description：Milvusと互換性がありますが、OceanBaseでは現時点で実用的な役割はありません
`def create_collection(self,collection_name: str,dimension: Optional[int] = None,primary_field_name: str = "id",id_type: Union[DataType, str] = DataType.INT64,vector_field_name: str = "vector",metric_type: str = "l2",auto_id: bool = False,timeout: Optional[float] = None,schema: Optional[CollectionSchema] = None, # Used for custom setupindex_params: Optional[IndexParams] = None, # Used for custom setupmax_length: int = 16384,**kwargs,)`	テーブルを作成します： collection_name : テーブル名 dimension : ベクトルデータの次元 primary_field_name: プライマリフィールド名 id_type: プライマリフィールドデータ型(VARCHARとINT型のみがサポートされています) vector_field_name : ベクトルフィールド名 metric_type: OceanBaseではまだ使用されていませんが、インターフェースの互換性は維持されています(主テーブルの定義においてベクトル距離関数を指定する必要がないため) auto_id：プライマリフィールドが自動インクリメントされるかどうか timeout : OceanBaseではまだ使用されていませんが、インターフェースの互換性は維持されています schema : カスタム集合アーキテクチャの場合、`schema` がNoneでない場合、上記のdimensionからmetric_typeまでのパラメータは無視されます index_params: カスタムベクトルインデックスパラメータ max_length: プライマリフィールドデータ型がVARCHARで、かつ `schema` がNoneでない場合の最大varcharの長さは `max_length`となります	`client.create_collection(collection_name=test_collection_name,schema=schema,index_params=idx_params,)`
`def get_collection_stats(self, collection_name: str, timeout: Optional[float] = None # pylint: disable=unused-argument) -> Dict:`	テーブルのレコードの数を取得します collection_name：テーブル名 timeout : OceanBaseではまだ使用されていませんが、インターフェースの互換性は維持されています
`def has_collection(self, collection_name: str, timeout: Optional[float] = None) -> bool`	テーブルが存在しているかどうかを判断します collection_name：テーブル名 timeout : OceanBaseではまだ使用されていませんが、インターフェースの互換性は維持されています
`def drop_collection(self, collection_name: str) -> None`	リネームテーブル old_name：元のテーブル名 new_name：新しいテーブル名
`def load_table(self, collection_name: str,)`	テーブルのメタデータをSQLAlchemyメタデータキャッシュに読み込みます collection_name：テーブル名

CollectionSchema & FieldSchema

MilvusLikeClientは、CollectionSchemaを使用してテーブルのモード定義を記述します。1つのCollectionSchemaは複数のFieldSchemaを含み、FieldSchemaは1つのテーブルの列モードを記述します。

MilvusLikeClientのcreate_schemaを使用してCollectionSchemaを作成します

def __init__(
    self,
    fields: Optional[List[FieldSchema]] = None,
    partitions: Optional[ObPartition] = None,
    description: str = "",  # ignored in oceanbase
    **kwargs,
)

パラメータの説明は以下のとおりです：

fields：FieldSchemaのオプションのセット。
partitions：パーティションルール(詳細についてはObPartitionを使用したパーティションルールの定義に関する章を参照してください)。
description：Milvusと互換性がありますが、OceanBaseでは現時点で実用的な役割はありません。

FieldSchemaを作成してCollectionSchemaに登録する

def add_field(self, field_name: str, datatype: DataType, **kwargs)

field_name：列名。
datatype：列データ型。サポートされているデータ型の詳細については、互換性に関する説明を参照してください。
kwargs：その他のパラメータは列属性の設定に使用されます。例は以下のとおりです：
```
def __init__(
self,
name: str,
dtype: DataType,
description: str = "",
is_primary: bool = False,
auto_id: bool = False,
nullable: bool = False,
**kwargs,
)
```
パラメータの説明は以下のとおりです：
- is_primary：主キーであるかどうか。
- auto_id：自動インクリメント列かどうか。
- nullable：空を許可するかどうか。

使用例

schema = self.client.create_schema()
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="title", datatype=DataType.VARCHAR, max_length=512)
schema.add_field(
    field_name="title_vector", datatype=DataType.FLOAT_VECTOR, dim=768
)
schema.add_field(field_name="link", datatype=DataType.VARCHAR, max_length=512)
schema.add_field(field_name="reading_time", datatype=DataType.INT64)
schema.add_field(
    field_name="publication", datatype=DataType.VARCHAR, max_length=512
)
schema.add_field(field_name="claps", datatype=DataType.INT64)
schema.add_field(field_name="responses", datatype=DataType.INT64)

self.client.create_collection(
    collection_name="medium_articles_2020", schema=schema
)

インデックス関連

API	パラメータ記述	例または備考
`def create_index(self,collection_name: str,index_params: IndexParams,timeout: Optional[float] = None,**kwargs,)`	作成済みのIndexParamsに基づいてベクトルインデックステーブルを作成する(このインターフェースのIndexParamsに関する使用の詳細については、prepare_index_paramsとadd_indexインターフェースを参照してください) collection_name：テーブル名 index_params：インデックスパラメータ timeout：OceanBaseではまだ使用されていませんが、インターフェースの互換性は維持されています kwargs：その他のパラメータについては、現在のところ使用されていませんが、互換性は維持されています
`def drop_index(self,collection_name: str,index_name: str,timeout: Optional[float] = None,**kwargs,)`	インデックステーブル削除 collection_name：テーブル名 index_name：インデックス名
`def refresh_index(self,collection_name: str,index_name: str,trigger_threshold: int = 10000,)`	読み取りの性能を向上させるためにベクトルインデックステーブルをリフレッシュします。これは増分データの移行とみなすことができます collection_name：テーブル名 index_name：インデックス名 trigger_threshold：リフレッシュ動作のトリガー閾値。インデックステーブルのデータ量がこの閾値を超えると、リフレッシュが行なわれます	OceanBaseで追加されたインターフェース Milvusとの互換性はありません
`def rebuild_index(self,collection_name: str,index_name: str,trigger_threshold: float = 0.2,)`	読み取りの性能を向上させるためにベクトルインデックステーブルを再構築します。これは増分データをベースラインインデックスデータにマージすることとみなすことができます collection_name：テーブル名 index_name：インデックス名 trigger_threshold：再構築動作のトリガー閾値。値の範囲は0から1で、増分データがフルデータに占める割合がこの閾値に達すると再構築がトリガーされます	OceanBaseに追加されたインターフェース Milvusとの互換性はありません
`def search(self,collection_name: str,data: list,anns_field: str,with_dist: bool = False,filter=None,limit: int = 10,output_fields: Optional[List[str]] = None,search_params: Optional[dict] = None,timeout: Optional[float] = None,partition_names: Optional[List[str]] = None,**kwargs,) -> List[dict]`	ベクトル近似近傍検索を実行します collection_name：テーブル名 data：検索が必要なベクトルデータ anns_field: 検索が必要なベクトル列名 with_dist: ベクトル距離の結果を返すかどうか filter : フィルタ条件付きベクトル近似近傍検索を使用します limit：top K output_fields: 出力列(または投影列) search_params : 値が `l2` / `neg_ip` の `metric_type` のみをサポートします（例：search_params = {"metric_type": "neg_ip"} ） timeout : OceanBaseでは使用されていません。互換性のためのみです partition_names : クエリを特定のパーティションに制限します戻り値：レコードのリストで、各レコードは辞典です column_nameから列値へのマッピングを表しています。	`res = self.client.search(collection_name=test_collection_name,data=[0, 0, 1],anns_field="embedding",limit=5,output_fields=["id"],search_params={"metric_type": "neg_ip"})self.assertEqual( set([r['id'] for r in res]), set([12, 111, 11, 112, 10]))`
`def query(self,collection_name: str,flter=None,output_fields: Optional[List[str]] = None,timeout: Optional[float] = None,partition_names: Optional[List[str]] = None,**kwargs,) -> List[dict]`	指定されたフィルタ条件を使用してデータレコードを読み取ります collection_name：テーブル名 flter : フィルタ条件付きのベクトル近似近傍検索を使用します output_fields: 出力列(または投影列) timeout : OceanBaseでは使用されていません。互換性のためのみです partition_names : クエリを特定のパーティションに制限します戻り値：レコードのリストで、各レコードは辞典です column_nameから列値へのマッピングを表しています。	`table = self.client.load_table(collection_name=test_collection_name)where_clause = [table.c["id"] < 100]res = self.client.query( collection_name=test_collection_name, output_fields=["id"], flter=where_clause,)`
`def get(self,collection_name: str,ids: Union[list, str, int],output_fields: Optional[List[str]] = None,timeout: Optional[float] = None,partition_names: Optional[List[str]] = None,**kwargs,) -> List[dict]`	指定された主キー `ids` のレコードを取得します： collection_name：テーブル名 ids：特定のidまたはidリスト。注意：MilvusLikeClient getインターフェースのidsパラメータは、ObVecClientのgetと異なります。詳細については、ObVecClient get を参照してください output_fields: 出力列(または投影列) timeout : OceanBaseでは使用されていません。互換性のためのみです partition_names : クエリを特定のパーティションに制限します戻り値：レコードのリストで、各レコードは辞典です column_nameから列値へのマッピングを表しています。	`res = self.client.get( collection_name=test_collection_name, output_fields=["id", "meta"], ids=[80, 12, 112],)`
`def delete(self,collection_name: str,ids: Optional[Union[list, str, int]] = None,timeout: Optional[float] = None, # pylint: disable=unused-argumentflter=None,partition_name: Optional[str] = "",**kwargs, # pylint: disable=unused-argument)`	集合内のデータを削除します collection_name：テーブル名 ids：特定のidまたはidリスト timeout : OceanBaseでは使用されていません。互換性のためのみです flter : フィルタ条件付きのベクトル近似近傍検索を使用します partition_name : 削除操作を特定のパーティションに制限します	`self.client.delete( collection_name=test_collection_name, ids=[12, 112], partition_name="p0")`
`def insert( self, collection_name: str, data: Union[Dict, List[Dict]], timeout: Optional[float] = None, partition_name: Optional[str] = "")`	テーブルにデータを挿入します collection_name：テーブル名 data：Key-Value形式で記述された挿入するデータ timeout : OceanBaseでは使用されていません。互換性のためのみです partition_name : 挿入操作を特定のパーティションに制限します	`data = [ {"id": 12, "embedding": [1, 2, 3], "meta": {"doc": "oceanbase document 1"}}, { "id": 90, "embedding": [0.13, 0.123, 1.213], "meta": {"doc": "oceanbase document 1"}, }, {"id": 112, "embedding": [1, 2, 3], "meta": None}, {"id": 190, "embedding": [0.13, 0.123, 1.213], "meta": None},]self.client.insert(collection_name=test_collection_name, data=data)`
`def upsert(self,collection_name: str,data: Union[Dict, List[Dict]],timeout: Optional[float] = None, # pylint: disable=unused-argumentpartition_name: Optional[str] = "",) -> List[Union[str, int]]`	テーブルのデータを更新します。主キーが重複した場合、それを交換します collection_name：テーブル名 data：更新挿入するデータの形式は、insertインターフェースと一致しています timeout : OceanBaseでは使用されていません。互換性のためのみです partition_name : 挿入操作を特定のパーティションに制限します	`data = [ {"id": 112, "embedding": [1, 2, 3], "meta": {'doc':'hhh1'}}, {"id": 190, "embedding": [0.13, 0.123, 1.213], "meta": {'doc':'hhh2'}},]self.client.upsert(collection_name=test_collection_name, data=data)`
`def perform_raw_text_sql(self, text_sql: str): return super().perform_raw_text_sql(text_sql)`	SQLステートメントを直接実行します text_sql: 実行するSQL 戻り値： SQLAlchemyが提供する結果セットのイテレータを返します

ObVecClient

コンストラクター

def __init__(
    self,
    uri: str = "127.0.0.1:2881",
    user: str = "root@test",
    password: str = "",
    db_name: str = "test",
    **kwargs,
)

テーブルモード関連操作

API	パラメータ記述	例または備考
`def check_table_exists(self, table_name: str)`	検査テーブルが存在しているかどうか table_name：テーブル名
`def create_table(self,table_name: str,columns: List[Column],indexes: Optional[List[Index]] = None,partitions: Optional[ObPartition] = None,)`	テーブル作成 table_name：テーブル名 columns：SQLAlchemyで定義されたテーブルの列モード indexes：SQLAlchemyで定義された一連のインデックステーブルモード partitions：オプションのパーティションルール(詳細については、ObPartitionを使用したパーティションルールの定義に関する章を参照してください)
`@classmethoddef prepare_index_params(cls)`	IndexParams対象を作成してベクトルインデックステーブルのモード定義を記録します `class IndexParams: """Vector index parameters for MilvusLikeClient" def init(self): self._indexes = {}` IndexParamsの定義は非常にシンプルで、内部には辞書タイプのメンバのみを持ち、 (列名、インデックス名)のtupleからIndexParam構造へのマッピングを保持します IndexParamタイプの構造関数は： `def init( self, index_name: str, field_name: str, index_type: Union[VecIndexType, str], kwargs)` index_name：ベクトルインデックステーブル名 field_name：ベクトル列名 index_type：ベクトルインデックスアルゴリズムタイプの列挙型で、現在はHNSWのみをサポートしています `prepare_index_params` を通じてIndexParamsを取得した後、`add_index` インターフェースを通じてIndexParamを登録できます： `def add_index( self, field_name: str, index_type: VecIndexType, index_name: str, kwargs)` パラメータの意味はIndexParamの構造関数と同じです	ベクトルインデックスを作成する使用例を示しています： `idx_params = self.client.prepare_index_params()idx_params.add_index( field_name="title_vector", index_type="HNSW", index_name="vidx_title_vector", metric_type="L2", params={"M": 16, "efConstruction": 256},)self.client.create_collection( collection_name=test_collection_name, schema=schema, index_params=idx_params,)` 注意が必要なのは `prepare_index_params` 関数はMilvusLikeClientで使用し、ObVecClientでは使用しないことが推奨されます。ObVecClientモードでは `create_index` インターフェースを使用してベクトルインデックステーブルを定義しなければなりません。(詳細については、create_indexインターフェースを参照してください)
`def create_table_with_index_params(self,table_name: str,columns: List[Column],indexes: Optional[List[Index]] = None,vidxs: Optional[IndexParams] = None,partitions: Optional[ObPartition] = None,)`	オプションのindex_paramsを使用して、テーブルを作成すると同時にベクトルインデックスを作成します table_name：テーブル名 columns：SQLAlchemyで定義されたテーブルの列モード indexes：SQLAlchemyで定義されたインデックステーブルの列モード vidxs：IndexParamsで指定されたベクトルインデックステーブルモード partitions：オプションのパーティションルール(詳細については、ObPartitionを使用したパーティションルールの定義に関する章を参照してください)	MilvusLikeClientでの使用が推奨されます。ObVecClientでの使用は推奨されません
`def create_index(self,table_name: str,is_vec_index: bool,index_name: str,column_names: List[str],vidx_params: Optional[str] = None,**kw,)`	通常インデックスとベクトルインデックスの2つのモードの作成をサポートします table_name：テーブル名 is_vec_index：インデックスまたはベクトルインデックスです index_name：インデックス名 column_names：どの列にインデックスを作成するか vidx_params：ベクトルインデックスのパラメータ。例：`“distance=l2, type=hnsw, lib=vsag”` 現在、OceanBaseは `type=hnsw` および `lib=vsag` のみをサポートしています。これらの2つの設定は保持してください。distanceは `l2` または `inner_product` に設定できます	`self.client.create_index( test_collection_name, is_vec_index=True, index_name="vidx", column_names=["embedding"], vidx_params="distance=l2, type=hnsw, lib=vsag",)`
`def create_vidx_with_vec_index_param(self,table_name: str,vidx_param: IndexParam,)`	ベクトルインデックスパラメータを使用してベクトルインデックスを作成します table_name：テーブル名 vidx_param：IndexParam構造のベクトルインデックスパラメータ
`def drop_table_if_exist(self, table_name: str)`	テーブルを削除します table_name：テーブル名
`def drop_index(self, table_name: str, index_name: str)`	インデックスを削除します table_name：テーブル名 index_name：インデックス名
`def refresh_index(self,table_name: str,index_name: str,trigger_threshold: int = 10000,)`	読み取りの性能を向上させるためにベクトルインデックステーブルをリフレッシュします。これは増分データの移行とみなすことができます table_name：テーブル名 index_name：インデックス名 trigger_threshold：リフレッシュ動作のトリガー閾値。インデックステーブルのデータ量がこの閾値を超えると、リフレッシュが行なわれます
`def rebuild_index(self,table_name: str,index_name: str,trigger_threshold: float = 0.2,)`	読み取りの性能を向上させるためにベクトルインデックステーブルを再構築します。これは増分データをベースラインインデックスデータにマージすることとみなすことができます table_name：テーブル名 index_name：インデックス名 trigger_threshold：再構築動作のトリガー閾値。値の範囲は0から1で、増分データがフルデータに占める割合がこの閾値に達すると再構築がトリガーされます

DML操作

API	パラメータ記述	例または備考
`def insert(self,table_name: str,data: Union[Dict, List[Dict]],partition_name: Optional[str] = "",)`	テーブルにデータを入力します table_name：テーブル名 data：Key-Value形式で記述された挿入するデータ partition_name：挿入操作を特定のパーティションに制限します	`vector_value1 = [0.748479, 0.276979, 0.555195]vector_value2 = [0, 0, 0]data1 = [{"id": i, "embedding": vector_value1} for i in range(10)]data1.extend([{"id": i, "embedding": vector_value2} for i in range(10, 13)])data1.extend([{"id": i, "embedding": vector_value2} for i in range(111, 113)])self.client.insert(test_collection_name, data=data1)`
`def upsert(self,table_name: str,data: Union[Dict, List[Dict]],partition_name: Optional[str] = "",)`	テーブルのデータを更新します。主キーが重複した場合、それを交換します。 table_name：テーブル名 data：更新挿入するデータ。key-value形式 partition_name：更新挿入を特定のパーティションに制限します
`def update(self,table_name: str,values_clause,where_clause=None,partition_name: Optional[str] = "",)`	テーブルのデータを更新します。主キーが重複した場合、それを交換します。 table_name：テーブル名 values_clause：更新列の値 where_clause：更新条件 partition_name：更新操作を特定のパーティションに制限します	`data = [ {"id": 112, "embedding": [1, 2, 3], "meta": {'doc':'hhh1'}}, {"id": 190, "embedding": [0.13, 0.123, 1.213], "meta": {'doc':'hhh2'}},]client.insert(collection_name=test_collection_name, data=data)client.update( table_name=test_collection_name, values_clause=[{'meta':{'doc':'HHH'}}], where_clause=[text("id=112")])`
`def delete(self,table_name: str,ids: Optional[Union[list, str, int]] = None,where_clause=None,partition_name: Optional[str] = "",)`	テーブルのデータを削除します table_name：テーブル名 ids：特定のidまたはidリスト where_clause：削除条件 partition_name：削除操作を特定のパーティションに制限します	`self.client.delete(test_collection_name, ids=["bcd", "def"])`
`def get(self,table_name: str,ids: Optional[Union[list, str, int]],where_clause = None,output_column_name: Optional[List[str]] = None,partition_names: Optional[List[str]] = None,)`	指定した主キー `ids` のレコードを取得します： table_name：テーブル名 ids：特定のidまたはidリスト。オプションのパラメータで、指定されていない場合は `ids=None` を入力できます。ObVecClient getインターフェースのidsパラメータはMilvusLikeClientのgetとは異なります。詳細については、MilvusLikeClient getを参照してください where_clause：取得条件 output_column_name：出力列または投影列の名前のセット partition_names：取得操作を特定のパーティションに制限します戻り値： MilvusLikeClientとは異なり、ObVecClientの戻り値はtuple listです。各tupleは1行のレコードを表します	`res = self.client.get( test_collection_name, ids=["abc", "bcd", "cde", "def"], where_clause=[text("meta->'$.page' > 1")], output_column_name=['id'])`
`def set_ob_hnsw_ef_search(self, ob_hnsw_ef_search: int)`	HNSWインデックスのefSearchパラメータを設定します。sessionレベル変数設定では、ef_searchが大きくなればリコール率も高くなるが、クエリの性能は若干下がります。 ob_hnsw_ef_search：HNSWインデックスのefSearchパラメータ
`def get_ob_hnsw_ef_search(self) -> int`	HNSWインデックスのefSearchパラメータを取得します
`def ann_search(self,table_name: str,vec_data: list,vec_column_name: str,distance_func,with_dist: bool = False,topk: int = 10,output_column_names: Optional[List[str]] = None,extra_output_cols: Optional[List] = None,where_clause=None,partition_names: Optional[List[str]] = None,**kwargs,)`	ベクトル近似近傍検索を実行します table_name：テーブル名 vec_data：検索が必要なベクトルデータ vec_column_name：検索が必要なベクトル列名 distance_func：距離関数。SQLAlchemy funcの拡張を提供し、`func.l2_distance`/`func.cosine_distance`/`func.inner_product`/`func.negative_inner_product` のオプションがあります。それぞれ、l2距離関数、cosine距離関数、内積距離関数、内積距離関数のマイナス値を表します with_dist：ベクトル距離付きの結果を返すかどうか topk：直近のどれくらいのベクトルを取得するか output_column_names：出力列または投影列の名前のセット extra_output_cols：追加の出力列。より複雑な出力式を提供できます where_clause：フィルタ条件 partition_names：クエリを特定のパーティションに制限します戻り値： MilvusLikeClientとは異なり、ObVecClientの戻り値はtuple listです。各tupleは1行のレコードを表します	`res = self.client.ann_search( test_collection_name, vec_data=[0, 0, 0], vec_column_name="embedding", distance_func=func.l2_distance, with_dist=True, topk=5, output_column_names=["id"],)`
`def precise_search(self,table_name: str,vec_data: list,vec_column_name: str,distance_func,topk: int = 10,output_column_names: Optional[List[str]] = None,where_clause=None,**kwargs,)`	正確近傍検索アルゴリズムを実行します table_name：テーブル名 vec_data：クエリのベクトル vec_column_name：ベクトル列名 distance_func：ベクトル距離関数。SQLAlchemy funcの拡張を提供し、func.l2_distance/func.cosine_distance/func.inner_product/func.negative_inner_productのオプションがあります。それぞれl2距離関数、cosine距離関数、内積距離関数、内積距離関数のマイナスの値を表します topk：直近のどれくらいのベクトルを取得するか output_column_names：出力列または投影列の名前のセット where_clause：フィルタ条件戻り値： MilvusLikeClientとは異なり、ObVecClientの戻り値はtuple listです。各tupleは1行のレコードを表します
`def perform_raw_text_sql(self, text_sql: str)`	SQLステートメントを直接実行します text_sql: 実行するSQL 戻り値： SQLAlchemyが提供する結果セットのイテレータを返します

ObPartitionを使用したパーティションルールの定義

pyobvectorは、range/range columns、list/list columns、hash、keyおよびサブパーティションをサポートするために以下のタイプを提供します：

ObRangePartition：rangeパーティション。構築する際に is_range_columns = True を設定してrange columnsパーティションを作成します。
ObListPartition：listパーティション。構築する際に is_list_columns = True を設定してlist columnsパーティションを作成します。
ObHashPartition：hashパーティション。
ObKeyPartition：keyパーティション。
ObSubRangePartition：rangeサブパーティション。構築する際に is_range_columns = True を設定してrange columnsサブパーティションを作成します。
ObSubListPartition：listサブパーティション。構築する際に is_list_columns = True を設定してlist columnsサブパーティションを作成します。
ObSubHashPartition：hashサブパーティション。
ObSubKeyPartition：keyサブパーティション。

rangeパーティションの例

range_part = ObRangePartition(
    False,
    range_part_infos=[
        RangeListPartInfo("p0", 100),
        RangeListPartInfo("p1", "maxvalue"),
    ],
    range_expr="id",
)

listパーティションの例

list_part = ObListPartition(
    False,
    list_part_infos=[
        RangeListPartInfo("p0", [1, 2, 3]),
        RangeListPartInfo("p1", [5, 6]),
        RangeListPartInfo("p2", "DEFAULT"),
    ],
    list_expr="col1",
)

hashパーティションの例

hash_part = ObHashPartition("col1", part_count=60)

マルチレベルパーティションの例

# rangeパーティション
range_columns_part = ObRangePartition(
    True,
    range_part_infos=[
        RangeListPartInfo("p0", 100),
        RangeListPartInfo("p1", 200),
        RangeListPartInfo("p2", 300),
    ],
    col_name_list=["col1"],
)
# rangeサブパーティション
range_sub_part = ObSubRangePartition(
    False,
    range_part_infos=[
        RangeListPartInfo("mp0", 1000),
        RangeListPartInfo("mp1", 2000),
        RangeListPartInfo("mp2", 3000),
    ],
    range_expr="col3",
)
range_columns_part.add_subpartition(range_sub_part)

純粋なSQLAlchemy APIモード

OceanBaseデータベースのベクトル検索機能で純粋なSQLAlchemy APIを使用したい場合には、以下の2つの方法を通じて同期データベースエンジンを取得できます：

方法1：ObVecClientを使用してデータベースエンジンの作成を補助する

from pyobvector import ObVecClient

client = ObVecClient(uri="127.0.0.1:2881", user="test@test")
engine = client.engine
# 次に、SQLAlchemyを通常通り使用してsessionを作成し、SQLAlchemyのAPIを使用します

方法2：ObVecClientの create_engine インターフェースを使用してデータベースエンジンを作成する

import pyobvector
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine

uri: str = "127.0.0.1:2881"
user: str = "root@test"
password: str = ""
db_name: str = "test"
registry.register("mysql.oceanbase", "pyobvector.schema.dialect", "OceanBaseDialect")
connection_str = (
    # mysql+aoceanbaseは、mysql標準を選択し、OceanBaseデータベースの同期ドライバーを使用することを意味します
    f"mysql+oceanbase://{user}:{password}@{uri}/{db_name}?charset=utf8mb4"
)
engine = create_engine(connection_str, **kwargs)
# 次に、SQLAlchemyを通常通り使用してsessionを作成し、SQLAlchemyのAPIを使用します

SQLAlchemyの非同期インターフェースを使用したい場合は、OceanBaseデータベースの非同期ドライバーを使用できます。

import pyobvector
from sqlalchemy.dialects import registry
from sqlalchemy.ext.asyncio import create_async_engine

uri: str = "127.0.0.1:2881"
user: str = "root@test"
password: str = ""
db_name: str = "test"
registry.register("mysql.aoceanbase", "pyobvector", "AsyncOceanBaseDialect")
connection_str = (
    # mysql+aoceanbaseは、mysql標準を選択し、OceanBaseデータベースの非同期ドライバーを使用することを意味します
    f"mysql+aoceanbase://{user}:{password}@{uri}/{db_name}?charset=utf8mb4"
)
engine = create_async_engine(connection_str)
# 次に、SQLAlchemyを通常通り使用してsessionを作成し、SQLAlchemyのAPIを使用します

さらなる例

pyobvectorコードリポジトリにアクセスして、より多くの例を取得します。

顧客事例

pyobvector Python SDKインターフェースの説明

MilvusLikeClient

コンストラクター

collection関連インターフェース

CollectionSchema & FieldSchema

MilvusLikeClientのcreate_schemaを使用してCollectionSchemaを作成します

FieldSchemaを作成してCollectionSchemaに登録する

使用例

インデックス関連

ObVecClient

コンストラクター

テーブルモード関連操作

DML操作

ObPartitionを使用したパーティションルールの定義

rangeパーティションの例

listパーティションの例

hashパーティションの例

マルチレベルパーティションの例

純粋なSQLAlchemy APIモード

さらなる例