CreateOpenVikingAccount 用于在指定的 OpenViking 企业版库下创建数据空间。
创建时,通过 ResourceID 指定所属库,通过 OpenVikingAccountID 指定新数据空间的名称。创建成功后,系统会在该数据空间下自动创建名为 default 的默认用户。
注意
本接口仅适用于企业版(enterprise)库,个人版(standard)库不支持该接口。OpenVikingAccountID 是调用者自定义的数据空间名称,不是火山引擎账号 ID,也不是 OpenViking 库 ID。
ResourceID。Action |
|
|---|---|
请求方法 | POST |
Content-Type |
|
字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
ResourceID | string | 是 | 数据空间所属的 OpenViking 库 ID。填写创建库接口返回的 |
OpenVikingAccountID | string | 是 | 要创建的数据空间名称,由调用者指定,长度为 1~64 个字符,后续操作该数据空间时使用此名称 |
字段 | 类型 | 说明 |
|---|---|---|
| bool | 创建操作是否成功, |
| string | 新建数据空间的名称,与请求中指定的值对应 |
HTTP 状态码 | 错误码 | 说明 |
|---|---|---|
400 |
| 请求参数格式错误或校验不通过 |
400 |
| 缺少必填参数 |
400 |
| 当前上下文不允许执行该请求 |
401 |
| 请求签名缺失或无效 |
403 |
| 身份校验失败 |
403 |
| 当前账号无权执行该操作 |
404 |
| 指定的库或资源不存在 |
429 |
| 超出配额或触发限流 |
500 |
| 服务内部错误 |
调用前,请替换 AK、SK、目标企业版库的 ResourceID,以及要创建的数据空间名称 OpenVikingAccountID。
import os import datetime import hashlib import hmac from urllib.parse import quote import requests, json def norm_query(params): query = "" for key in sorted(params.keys()): if type(params[key]) == list: for k in params[key]: query = ( query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&" ) else: query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&") query = query[:-1] return query.replace("+", "%20") def hmac_sha256(key: bytes, content: str): return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest() def hash_sha256(content: str): return hashlib.sha256(content.encode("utf-8")).hexdigest() class ClientForConsoleApi: def __init__(self, ak, sk, host, region): self.ak = ak self.sk = sk self.host = host self.region = region self.service_code = "vikingdb" self.version = "2025-06-09" def request(self, method, action, body, query=None, header=None): if query is None: query = {} if header is None: header = {} credential = { "access_key_id": self.ak, "secret_access_key": self.sk, "service": self.service_code, "region": self.region, } request_param = { "body": json.dumps(body), "host": self.host, "path": "/", "method": method, "content_type": 'application/json', "date": datetime.datetime.utcnow(), "query": {"Action": action, "Version": self.version, **query}, } if body is None: request_param["body"] = "" x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ") short_x_date = x_date[:8] x_content_sha256 = hash_sha256(request_param["body"]) sign_result = { "Host": request_param["host"], "X-Content-Sha256": x_content_sha256, "X-Date": x_date, "Content-Type": request_param["content_type"], } signed_headers_str = ";".join( ["content-type", "host", "x-content-sha256", "x-date"] ) canonical_request_str = "\n".join( [request_param["method"].upper(), request_param["path"], norm_query(request_param["query"]), "\n".join( [ "content-type:" + request_param["content_type"], "host:" + request_param["host"], "x-content-sha256:" + x_content_sha256, "x-date:" + x_date, ] ), "", signed_headers_str, x_content_sha256, ] ) hashed_canonical_request = hash_sha256(canonical_request_str) credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"]) string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request]) k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date) k_region = hmac_sha256(k_date, credential["region"]) k_service = hmac_sha256(k_region, credential["service"]) k_signing = hmac_sha256(k_service, "request") signature = hmac_sha256(k_signing, string_to_sign).hex() sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format( credential["access_key_id"] + "/" + credential_scope, signed_headers_str, signature, ) header = {**header, **sign_result} r = requests.request(method=method, url="https://{}{}".format(request_param["host"], request_param["path"]), headers=header, params=request_param["query"], data=request_param["body"], ) return r.status_code, r.headers, r.json() if __name__ == '__main__': client = ClientForConsoleApi( ak = "*",#替换为您的 AK sk ="*" ,#替换为您的 SK host="vikingdb.cn-beijing.volcengineapi.com", region="cn-beijing" ) http_code, _, result_json = client.request( method="POST", action="CreateOpenVikingAccount", body={ "ResourceID": "ov-xxx", "OpenVikingAccountID": "space_a" } ) print("req http status code: ", http_code) print("req result: \n", result_json)
{ "ResponseMetadata": { "RequestId": "202609151030001A2B3C4D5E6F", "Action": "CreateOpenVikingAccount", "Version": "2025-06-09", "Service": "vikingdb", "Region": "cn-beijing" }, "Result": { "Success": true, "OpenVikingAccountID": "space_a" } }