pyobvector Python SDK interface|V4.4.2| docs|Distributed Database

pyobvector Python SDK interface

Last Updated：2026-03-06 07:02:41 Updated

pyobvector is the Python SDK for OceanBase's vector storage functionality. It provides two usage modes:

pymilvus compatible mode: Uses the MilvusLikeClient object to interact with the database, offering commonly used interfaces compatible with the lightweight MilvusClient.
SQLAlchemy extension mode: Uses the ObVecClient object to interact with the database, providing an extension to the Python SDK for relational databases.

This topic describes the interfaces and provides examples for both modes.

MilvusLikeClient

Constructor

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

API	Description	Example
`def create_schema(self, **kwargs) -> CollectionSchema:`	Constructs a CollectionSchema object Parameters are optional. If not provided, an empty schema is created. Optional parameters include: fields: A list of FieldSchema objects (see the add_schema API for details) partitions: Partitioning rules (see the section on defining partitioning rules using ObPartition) description: This is compatible with Milvus but has no effect in 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 setup index_params: Optional[IndexParams] = None, # Used for custom setup max_length: int = 16384, **kwargs, )`	Creates a table: collection_name: The name of the table dimension: The dimension of the vector data primary_field_name: The name of the primary field id_type: The data type of the primary field (only VARCHAR and INT are supported) vector_field_name: The name of the vector field metric_type: Not used in OceanBase but retained for interface compatibility (since the primary table definition does not require specifying a vector distance function) auto_id: Whether the primary field is auto-incremented timeout: Not used in OceanBase but retained for interface compatibility schema: Custom collection schema. If provided, the parameters from dimension to metric_type will be ignored index_params: Custom vector index parameters max_length: The maximum length of the VARCHAR primary field when `schema` is provided	`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:`	Retrieves the number of records in the table collection_name: The name of the table timeout: Not used in OceanBase but retained for interface compatibility
`def has_collection(self, collection_name: str, timeout: Optional[float] = None) -> bool`	Checks if the table exists collection_name: The name of the table timeout: Not used in OceanBase but retained for interface compatibility
`def drop_collection(self, collection_name: str) -> None`	Renames the table old_name: The original name of the table new_name: The new name of the table
`def load_table(self, collection_name: str,)`	Loads table metadata into the SQLAlchemy metadata cache collection_name: The name of the table

CollectionSchema & FieldSchema

MilvusLikeClient uses CollectionSchema to describe the schema of a table. A CollectionSchema contains multiple FieldSchema, each of which describes the schema of a column in the table.

Create a CollectionSchema using MilvusLikeClient

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

The parameters are described as follows:

fields: A list of optional FieldSchema objects.
partitions: The partitioning rules (see the section on defining partitioning rules using ObPartition for details).
description: This is for compatibility with Milvus and has no effect in OceanBase.

Create a FieldSchema and register it to a CollectionSchema

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

field_name: The name of the column.
datatype: The data type of the column (see Compatibility for supported data types).
kwargs: Additional parameters for configuring column properties, as follows:
```
def __init__(
self,
name: str,
dtype: DataType,
description: str = "",
is_primary: bool = False,
auto_id: bool = False,
nullable: bool = False,
**kwargs,
)
```
The parameters are described as follows:
- is_primary: Whether the column is a primary key.
- auto_id: Whether the column is an auto-increment column.
- nullable: Whether the column allows null values.

Example

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	Description	Example or Note
`def create_index( self, collection_name: str, index_params: IndexParams, timeout: Optional[float] = None, **kwargs, )`	Creates a vector index table based on the constructed IndexParams (for more details on using IndexParams, see the prepare_index_params and add_index interfaces). collection_name: the name of the table index_params: the index parameters timeout: not used in OceanBase, but kept for interface compatibility kwargs: other parameters, currently not used, kept for compatibility
`def drop_index( self, collection_name: str, index_name: str, timeout: Optional[float] = None, **kwargs, )`	Drops an index table. collection_name: the name of the table index_name: the name of the index
`def refresh_index( self, collection_name: str, index_name: str, trigger_threshold: int = 10000, )`	Refreshes the vector index table to improve read performance, which can be understood as migrating incremental data. collection_name: the name of the table index_name: the name of the index trigger_threshold: the threshold for triggering a refresh. If the data volume of the index table exceeds this threshold, a refresh will be performed.	OceanBase's additional interface Not compatible with Milvus
`def rebuild_index( self, collection_name: str, index_name: str, trigger_threshold: float = 0.2, )`	Rebuilds the vector index table to improve read performance, which can be understood as merging incremental data into the base index data. collection_name: the name of the table index_name: the name of the index trigger_threshold: the threshold for triggering a rebuild. The value ranges from 0 to 1. A rebuild is triggered when the ratio of incremental data to total data reaches this threshold.	OceanBase's additional interface Not compatible with 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]`	Executes a vector approximate nearest neighbor search. collection_name: the name of the table data: the vector data to be searched anns_field: the name of the vector column to be searched with_dist: whether to return results with vector distances filter: performs a vector approximate nearest neighbor search with a filter condition limit: top K output_fields: output columns (also known as projection columns) search_params: only supports the `metric_type` value of `l2`/`neg_ip` (e.g., search_params = {"metric_type": "neg_ip"}). timeout: not used in OceanBase, only for compatibility partition_names: restricts the query to certain partitions Return value: A list of records, each of which is a dictionary Mapping from column name to column value.	`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]`	Reads data records using the specified filter condition. collection_name: the name of the table flter: performs a vector approximate nearest neighbor search with a filter condition output_fields: output columns (also known as projection columns) timeout: not used in OceanBase, only for compatibility partition_names: restricts the query to certain partitions Return value: A list of records, each of which is a dictionary Mapping from column name to column value.	`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]`	Retrieves records with the specified primary key `ids`: collection_name: the name of the table ids: a single id or a list of ids. Note: The ids parameter in the MilvusLikeClient get interface is different from that in the ObVecClient get interface. For more details, see ObVecClient get output_fields: output columns (also known as projection columns) timeout: not used in OceanBase, only for compatibility partition_names: restricts the query to certain partitions Return value: A list of records, each of which is a dictionary Mapping from column name to column value.	`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-argument flter=None, partition_name: Optional[str] = "", **kwargs, # pylint: disable=unused-argument )`	Deletes data from the collection. collection_name: the name of the table ids: a single id or a list of ids timeout: not used in OceanBase, only for compatibility flter: performs a vector approximate nearest neighbor search with a filter condition partition_name: restricts the delete operation to a specific partition	`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] = "" )`	Inserts data into the table. collection_name: the name of the table data: the data to be inserted, described in key-value format timeout: not used in OceanBase, only for compatibility partition_name: restricts the insert operation to a specific partition	`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-argument partition_name: Optional[str] = "", ) -> List[Union[str, int]]`	Updates data in the table. If the primary key exists, the corresponding record is updated; otherwise, a new record is inserted. collection_name: the name of the table data: the data to be inserted or updated, with the same format as the insert interface timeout: not used in OceanBase, only for compatibility partition_name: restricts the operation to a specific partition	`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)`	Executes an SQL statement directly. text_sql: the SQL statement to be executed Return value: Returns an iterator of result sets provided by SQLAlchemy

ObVecClient

Constructor

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

Table mode operations

API	Description	Example or Notes
`def check_table_exists(self, table_name: str)`	Checks if a table exists table_name: The name of the table.
`def create_table( self, table_name: str, columns: List[Column], indexes: Optional[List[Index]] = None, partitions: Optional[ObPartition] = None, )`	Creates a table table_name: The name of the table. columns: The column schema of the table defined using SQLAlchemy. indexes: A list of index schemas defined using SQLAlchemy. partitions: Optional partitioning rules (see the section on defining partitioning rules using ObPartition for details).
`@classmethod def prepare_index_params(cls)`	Creates an IndexParams object to record the schema definition of the vector index table `class IndexParams: """Vector index parameters for MilvusLikeClient" def init(self): self._indexes = {}` IndexParams is defined very simply, with only one dictionary member internally storing a mapping from (column name, index name) tuples to IndexParam structures. The constructor for the IndexParam class is `def init( self, index_name: str, field_name: str, index_type: Union[VecIndexType, str], kwargs )` index_name: The name of the vector index table. field_name: The name of the vector column. index_type: An enumeration class representing the type of vector index algorithm. Currently, only HNSW is supported. After obtaining an IndexParams object using `prepare_index_params`, you can register an IndexParam using the `add_index` interface: `def add_index( self, field_name: str, index_type: VecIndexType, index_name: str, kwargs )` The parameters have the same meaning as those in the constructor of IndexParam.	Here's an example of creating a vector index: `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, )` Note that the `prepare_index_params` function is recommended to be used under MilvusLikeClient and not under ObVecClient. In ObVecClient mode, the `create_index` interface should be used to define vector index tables. (See the create_index interface for details.)
`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, )`	Creates a table and simultaneously creates vector indexes using optional index_params table_name: The name of the table. columns: The column schema of the table defined using SQLAlchemy. indexes: A list of index schemas defined using SQLAlchemy. vidxs: The vector index schema specified by IndexParams. partitions: Optional partitioning rules (see the section on defining partitioning rules using ObPartition for details).	Recommended to use under MilvusLikeClient, not recommended to use under 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, )`	Supports creating both regular indexes and vector indexes table_name: The name of the table. is_vec_index: Indicates whether it is a regular index or a vector index. index_name: The name of the index. column_names: The columns on which to create the index. vidx_params: Vector index parameters, for example, `"distance=l2, type=hnsw, lib=vsag"` Currently, OceanBase only supports `type=hnsw` and `lib=vsag`. Please retain these settings. The distance can be set to `l2` or `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, )`	Creates a vector index using vector index parameters table_name: The name of the table. vidx_param: The vector index parameters constructed by IndexParam.
`def drop_table_if_exist(self, table_name: str)`	Drops a table table_name: The name of the table.
`def drop_index(self, table_name: str, index_name: str)`	Drops an index table_name: The name of the table. index_name: The name of the index.
`def refresh_index( self, table_name: str, index_name: str, trigger_threshold: int = 10000, )`	Refreshes the vector index table to improve read performance, which can be understood as the migration of incremental data table_name: The name of the table. index_name: The name of the index. trigger_threshold: The threshold for triggering a refresh. If the data volume of the index table exceeds this threshold, a refresh is performed.
`def rebuild_index( self, table_name: str, index_name: str, trigger_threshold: float = 0.2, )`	Rebuilds the vector index table to improve read performance, which can be understood as merging incremental data into the base index data table_name: The name of the table. index_name: The name of the index. trigger_threshold: The threshold for triggering a rebuild, ranging from 0 to 1. A rebuild is triggered when the proportion of incremental data to total data reaches this threshold.

DML operations

API	Parameter Description	Example or Remarks
`def insert( self, table_name: str, data: Union[Dict, List[Dict]], partition_name: Optional[str] = "", )`	Insert data into the table table_name: The name of the table. data: The data to be inserted, in key-value format. partition_name: The partition to which the insertion is restricted.	`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] = "", )`	Insert or update data in the table. If the primary key exists, the corresponding record is updated; otherwise, a new record is inserted. table_name: The name of the table. data: The data to be inserted or updated, in key-value format. partition_name: The partition to which the operation is restricted.
`def update( self, table_name: str, values_clause, where_clause=None, partition_name: Optional[str] = "", )`	Update data in the table. If the primary key is duplicated, it will be replaced. table_name: The name of the table. values_clause: The values to update for the columns. where_clause: The update conditions. partition_name: The partitions to which the update operation is restricted.	`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] = "", )`	Delete data from the table table_name: The name of the table. ids: A single id or a list of ids. where_clause: The deletion conditions. partition_name: The partitions to which the deletion operation is restricted.	`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, )`	Retrieve records with the specified primary key `ids`. table_name: The name of the table. ids: A single id or a list of ids. Optional; if not provided, set `ids=None`. Note that the `ids` parameter in the ObVecClient get interface differs from that in MilvusLikeClient. For more information, see MilvusLikeClient get. where_clause: The retrieval conditions. output_column_name: A list of output columns or projected column names. partition_names: The partitions to which the retrieval operation is restricted. Return Value: Unlike MilvusLikeClient, the return value of ObVecClient is a list of tuples, where each tuple represents a record.	`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)`	Set the efSearch parameter for the HNSW index. This is a session-level variable. A higher ef_search value improves recall but decreases query performance. ob_hnsw_ef_search: The efSearch parameter for the HNSW index.
`def get_ob_hnsw_ef_search(self) -> int`	Retrieve the efSearch parameter for the HNSW index
`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, )`	Perform approximate nearest neighbor search table_name: The name of the table. vec_data: The vector data to be searched. vec_column_name: The name of the vector column to be searched. distance_func: The distance function. This supports SQLAlchemy func extensions, with optional values including `func.l2_distance`, `func.cosine_distance`, `func.inner_product`, and `func.negative_inner_product`, representing L2 distance, cosine distance, inner product, and negative inner product, respectively. with_dist: Whether to return results with vector distances. topk: The number of nearest vectors to retrieve. output_column_names: A list of output columns or projected column names. extra_output_cols: Additional output columns, allowing for more complex output expressions. where_clause: The filtering conditions. partition_names: The partitions to which the query is restricted. Return Value: Unlike MilvusLikeClient, the return value of ObVecClient is a list of tuples, where each tuple represents a record.	`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, )`	Perform precise nearest neighbor search table_name: The name of the table. vec_data: The query vector. vec_column_name: The name of the vector column. distance_func: The vector distance function, supporting SQLAlchemy func extensions, with optional values including func.l2_distance, func.cosine_distance, func.inner_product, and func.negative_inner_product, representing L2 distance, cosine distance, inner product, and negative inner product, respectively. topk: The number of nearest vectors to retrieve. output_column_names: A list of output columns or projected column names. where_clause: The filtering conditions. Return Value: Unlike MilvusLikeClient, the return value of ObVecClient is a list of tuples, where each tuple represents a record.
`def perform_raw_text_sql(self, text_sql: str)`	Execute an SQL statement directly text_sql: The SQL statement to be executed. Return Value: Returns an iterator of result sets provided by SQLAlchemy

Define partitioning rules using ObPartition

pyobvector provides the following types to support range/range columns, list/list columns, hash, key, and subpartitioning:

ObRangePartition: A range partition. Set is_range_columns = True during construction to create a range columns partition.
ObListPartition: A list partition. Set is_list_columns = True during construction to create a list columns partition.
ObHashPartition: A hash partition.
ObKeyPartition: A key partition.
ObSubRangePartition: A subpartition for range partitioning. Set is_range_columns = True during construction to create a range columns subpartition.
ObSubListPartition: A subpartition for list partitioning. Set is_list_columns = True during construction to create a list columns subpartition.
ObSubHashPartition: A subpartition for hash partitioning.
ObSubKeyPartition: A subpartition for key partitioning.

Example of range partitioning

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

Example of list partitioning

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

Example of hash partitioning

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

Example of multi-level partitioning

# Primary range partition
range_columns_part = ObRangePartition(
    True,
    range_part_infos=[
        RangeListPartInfo("p0", 100),
        RangeListPartInfo("p1", 200),
        RangeListPartInfo("p2", 300),
    ],
    col_name_list=["col1"],
)
# Subpartition for range partitioning
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)

Pure SQLAlchemy API mode

If you want to use the vector search feature of OceanBase Database with the pure SQLAlchemy API, you can obtain the synchronous database engine in the following two ways:

Method 1: Use ObVecClient to create the database engine.

from pyobvector import ObVecClient

client = ObVecClient(uri="127.0.0.1:2881", user="test@test")
engine = client.engine
# You can then proceed to create a session using SQLAlchemy and use the SQLAlchemy API.

Method 2: Use the create_engine interface of ObVecClient to create the database 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+oceanbase indicates selecting the mysql standard and using the synchronous driver of OceanBase Database.
    f"mysql+oceanbase://{user}:{password}@{uri}/{db_name}?charset=utf8mb4"
)
engine = create_engine(connection_str, **kwargs)
# You can then proceed to create a session using SQLAlchemy and use the SQLAlchemy API.

If you want to use the asynchronous interface of SQLAlchemy, you can use the asynchronous driver of OceanBase Database:

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 indicates selecting the mysql standard and using the asynchronous driver of OceanBase Database.
    f"mysql+aoceanbase://{user}:{password}@{uri}/{db_name}?charset=utf8mb4"
)
engine = create_async_engine(connection_str)
# You can then proceed to create a session using SQLAlchemy and use the SQLAlchemy API.

More examples

For more examples, visit the pyobvector code repository.

pyobvector Python SDK interface

MilvusLikeClient

Constructor

Collection related interfaces

CollectionSchema & FieldSchema

Create a CollectionSchema using MilvusLikeClient

Create a FieldSchema and register it to a CollectionSchema

Example

Index related

ObVecClient

Constructor

Table mode operations

DML operations

Define partitioning rules using ObPartition

Example of range partitioning

Example of list partitioning

Example of hash partitioning

Example of multi-level partitioning

Pure SQLAlchemy API mode

More examples