Python API

from transwarp_hippo_api.hippo_client import *
  
if __name__ == "__main__":
    # Connect to Hippo
    hc = HippoClient(["ip:port"])
 
    # Create database
    # db_created = hc.create_database("my_database")
 
    # List database
    # print(hc.list_databases())
 
    # Delete database
    # hc.delete_database("my_database")
 
    # Create table
    # tbl_name = "my_test"
 
    # Declare field
    # fields = [
    #     HippoField("pk", True, HippoType.INT64),
    #     HippoField("vector", False, HippoType.FLOAT_VECTOR, type_params = {"dimension": 2}),
    #     HippoField("text", False, HippoType.STRING),
    #     HippoField("number", False, HippoType.INT64)
    # ]
    # hc.create_table(name = tbl_name, database_name = "my_database", fields = fields, number_of_shards = 1, number_of_replicas = 1)
  
    # Rename table
    # print(hc.rename_table("my_test", "my_test_rename", "my_database"))
  
    # Delete table
    # hc.delete_table("my_test_rename", "my_database")
  
    # Check table exists
    # print(hc.check_table_exists("my_test_rename", "my_database"))
  
    # Get table details
    # print(hc.get_table_info("my_test_rename", "my_database"))
  
    # Get table schema
    # print(hc.get_table_schema("my_test_rename", "my_database"))
  
    # Get table vector index
    # print(hc.get_table_indexes("my_test_rename", "my_database"))
  
    # List tables of database
    # print(hc.list_tables("my_database", pattern="my_test_rename"))
  
    # List shards of table
    # print(hc.list_shards("my_test_rename", "my_database"))
  
    # List table configuration
    # print(hc.get_table_config("my_test_rename", "my_database"))
  
    # Update table configuration
    # hc.update_table_config("my_test_rename", data_center = "DEFAULT", embedding_segment_seal_proportion = 0.3, database_name = "my_database")
  
    # Create alias
    # hc.create_alias("my_test_rename", "my_test_rename_alias", "my_database")
 
    # Delete alias
    # hc.delete_alias("my_test_rename_alias", "my_database")
  
    # Get table
    yy = hc.get_table("my_test_rename", "my_database")
  
    # Insert
    # yy.insert_rows(
    #     [
    #         [3, 4, 5],
    #         [[-0.1, -0.5], [0.7, 0.1], [0.1, -0.2]],
    #         ["你好吗", "大家好吗", "hello"],
    #         [100, 200, 300]
    #     ],
    # )
  
    # Delete
    # yy.delete_rows([[1234], [None], [None], [None]])
  
    # Update
    # yy.update_rows([[3], [[0.4, 0.5]], ["?"], [101]])
    # yy.upsert_rows([[3], [[0.4, 0.5]], ["?"], [101]])
  
    # Copy an existing table
    # hc.copy_table("my_test_rename", "my_test_copy", "my_database", "default")
  
    # Batch delete
    # yy.delete_rows_by_query(expr = "number < 200")
  
    # Create vector index
    # yy.create_index(field_name = "vector", index_name = "vector_index",
    #                 index_type = IndexType.IVF_FLAT, metric_type = MetricType.L2,
    #                 nlist = 10, )
  
    # Activate vector index
    # yy.activate_index("vector_index")
  
    # Release vector index
    # yy.release_index("vector_index")
  
    # Load vector index
    # yy.load_index("vector_index")
  
    # Get vector index information
    # print(yy.get_index())
  
    # Drop index
    # yy.drop_index("vector_index")
  
    # Create scalar index
    # yy.create_scalar_index(field_names = ["text"], index_name = "text1_index")
  
    # Delete scalar index
    # yy.delete_scalar_index(index_name = "text1_index")
  
    # Vector similarity search
    # result = yy.query(
    #     "vector",
    #     [[0.9, 0.5]],
    #     ["pk", "text", "vector", "number"],
    #     3
    # )
    # print(result)
  
    # Hybrid search
    # result = yy.query(
    #     "vector",
    #     [[0.9, 0.5]],
    #     ["pk", "text", "vector", "number"],
    #     3,
    #     dsl = "number > 200"
    # )
    # print(result)
  
    # Scalar search
    # result = yy.query_scalar(
    #     ["pk", "text", "vector", "number"],
    #     expr = "number > 200"
    # )
    # print(result)
  
    # Create user
    # hc.create_user(
    #     user_name = "test_user",
    #     pwd = "password123"
    # )
  
    # Get user info
    # print(hc.get_user_info("test_user"))
  
    # Delete user
    # hc.delete_user("test_user")
  
    # Change user configuration
    # hc.change_user_info(user_name = "test_user", pwd = "password123", is_super = True)
  
    # Change user password
    # hc.change_password(new_password = "new_password123", user_name = "test_user")
  
    # Grant permission
    # hc.grant_user_permission("test_user", ["all"], "my_test_rename", "my_database")
  
    # Delete permission
    # hc.delete_user_permission("test_user", ["all"], "my_test_rename", "my_database")
  
    # Check user permission
    # print(hc.view_user_permission("test_user"))
  
    # Get job
    # print(hc.get_job())
  
    # Delete job
    # hc.delete_job("")
  
    # Check table in trash
    # print(hc.view_tables_in_trash("my_database"))
  
    # Delete table in trash
    # hc.delete_table_in_trash("my_test_rename", "my_database")