接口用于在指定的数据集 Collection 内写入数据。指定写入的数据是一个map，允许单次插入一条数据或者多条数据，单次最多可插入100条数据。

• 对应的V1接口为：https://www.volcengine.com/docs/84313/1254533
• 使用区别：

• 注：QPS限流是以数据条数计算，V1与V2接口的限流行为完全相同。

创建时间类型字段(date_time)

只能填写以下格式的其中一种，全部遵循RFC3339标准（https://datatracker.ietf.org/doc/html/rfc3339）

公共响应体参数部分见
数据面API调用流程

1.写入带直接向量字段的数据集

• 请求参数

req_path = "/api/vikingdb/data/upsert"
req_body = {
    "collection_name": "test_coll",
    "data": [
        {
            "f_id": "000135", #以下参数为选填，请根据coll创建字段填写
            "f_vector": [0.1, 0.33, -0.88, 0.66],
            "f_city": "北京"
        }
    ]
}

• 响应参数

{
    "code": "Success",
    "message": "The API call was executed successfully.",
    "request_id": "02175438839168500000000000000000000ffff0a003ee4fc3499",
    "result": null
}

2.写入带向量化字段的数据集

图文示例

• 请求参数

req_path = "/api/vikingdb/data/upsert"
req_body = {
    "collection_name": "test_coll_with_vectorize",
    "data": [
        {
            "f_id": "000135",
            "f_city": "北京",
            "f_text": "这是一件历史悠久的文物，具有1300年的历史",
            "f_image": "tos://my_bucket/good_000135.jpg"
        }
    ]
}

• 响应参数

{
    "code": "Success",
    "message": "The API call was executed successfully.",
    "request_id": "02175438839168500000000000000000000ffff0a003ee4fc3499",
    "result": {
        "token_usage": {
            "doubao-embedding-vision__250615": {
                "prompt_tokens": 1325,
                "completion_tokens": 0,
                "image_tokens": 1312,
                "total_tokens": 1325
            }
        }
    }
}

视频示例

• 请求参数

req_path = "/api/vikingdb/data/upsert"
req_body = {
    "collection_name": "test_coll_with_vectorize",
    "data": [
        {
            "f_id": "000135",
            "f_city": "北京",
            "f_text": "这是一件历史悠久的文物，具有1300年的历史",
            "f_image": "tos://my_bucket/good_000135.jpg",
            "f_video": {
                "value": "tos://my_bucket/good_000135.mp4",
                "fps": 2.0,
             }
        }
    ]
}

• 响应参数

{
    "code": "Success",
    "message": "The API call was executed successfully.",
    "request_id": "02175438839168500000000000000000000ffff0a003ee4fc3499",
    "result": {
        "token_usage": {
            "doubao-embedding-vision__250615": {
                "prompt_tokens": 1325,
                "completion_tokens": 0,
                "image_tokens": 1312,
                "total_tokens": 1325
            }
        }
    }
}

3. 写入时间、地理字段

写入时间类型示例

req_path = "/api/vikingdb/data/upsert"
req_body = {
    "collection_name": "test",
    "data": [
        {"f_id": 1, "f_time": "2025-08-12T12:34:56+08:00", "f_vector":[...]},
        {"f_id": 2, "f_time": "2025-08-12T11:33:56+08:00", "f_vector":[...]},
        {"f_id": 3, "f_time": "2025-08-12T04:32:56+08:00", "f_vector":[...]}
    ]
}

写入地理位置类型示例

req_path = "/api/vikingdb/data/upsert"
req_body = {
    "collection_name": "test",
    "data": [
        {"f_id": 1, "f_geo_point": "116.403874,39.914885", "f_vector":[...]},
        {"f_id": 2, "f_geo_point": "116.412138,39.914912", "f_vector":[...]},
        {"f_id": 3, "f_geo_point": "116.408108,39.915023", "f_vector":[...]}
    ]
}

"""
pip3 install volcengine
"""
import os

from volcengine.auth.SignerV4 import SignerV4
from volcengine.Credentials import Credentials
from volcengine.base.Request import Request
import requests, json

class ClientForDataApi:
    def __init__(self, ak, sk, host):
        self.ak = ak
        self.sk = sk
        self.host = host

    def prepare_request(self, method, path, params=None, data=None):
        r = Request()
        r.set_shema("https")
        r.set_method(method)
        r.set_connection_timeout(10)
        r.set_socket_timeout(10)
        mheaders = {
            'Accept': 'application/json',
            'Content-Type': 'application/json',
            'Host': self.host,
        }
        r.set_headers(mheaders)
        if params:
            r.set_query(params)
        r.set_host(self.host)
        r.set_path(path)
        if data is not None:
            r.set_body(json.dumps(data))
        credentials = Credentials(self.ak, self.sk, 'vikingdb', 'cn-beijing')
        SignerV4.sign(r, credentials)
        return r
        
    def do_req(self, req_method, req_path, req_params, req_body):
        req = self.prepare_request(method=req_method, path=req_path, params=req_params, data=req_body)
        return requests.request(method=req.method, url="http://{}{}".format(self.host, req.path),
                                  headers=req.headers, data=req.body, timeout=10000)

if __name__ == '__main__':
    client = ClientForDataApi(
        ak = "*",#替换为您的ak
        sk = "*",#替换为您的sk
        host = "api-vikingdb.vikingdb.cn-beijing.volces.com",#替换为您所在的域名
    )
    req_method = "POST"
    req_params = None
    req_path = "/api/vikingdb/data/upsert"
    req_body = {
        "collection_name": "test_coll_with_vectorize",#替换为您的collection名
        "data": [
            {
                "f_id": "000135",
                "f_city": "北京",
                "f_text": "这是一件历史悠久的文物，具有1300年的历史",
                "f_image": "tos://my_bucket/good_000135.jpg"
            }
        ]
    }
    result = client.do_req(req_method=req_method, req_path=req_path, req_params=req_params, req_body=req_body)
    print("req http status code: ", result.status_code)
    print("req result: \n", result.text)