ListOpenVikingAccounts 用于分页查询指定 OpenViking 企业版库下的数据空间,支持按数据空间名称进行模糊搜索。
返回信息包括数据空间名称、用户数量、创建时间,以及是否为默认数据空间。
注意
本接口仅适用于企业版(enterprise)库。
不传 Keyword 时,分页列出库下的数据空间;传入 Keyword 时,按数据空间名称进行模糊搜索。
ResourceID。Action |
|
|---|---|
请求方法 | POST |
Content-Type |
|
字段 | 类型 | 必填 | 填写说明 |
|---|---|---|---|
| string | 是 | 待查询的 OpenViking 库 ID,可从库信息页面获取,不能填写库名称 |
| string | 否 | 数据空间名称的搜索关键词,支持模糊搜索 |
| int | 否 | 查询页码,默认 |
| int | 否 | 每页返回条数,默认 |
成功响应包含 ResponseMetadata 和 Result 两部分。以下字段位于 Result 中。
字段 | 类型 | 说明 |
|---|---|---|
| array | 当前页的数据空间列表,元素结构见下表 |
| int | 数据空间总数,用于分页展示 |
AccountList 中的每个元素包含以下字段:
字段 | 类型 | 说明 |
|---|---|---|
| string | 数据空间名称,例如 |
| int | 该数据空间下的用户数量 |
| string | 创建时间。以字符串形式返回时间戳,例如 |
| bool | 是否为默认数据空间。 |
HTTP 状态码 | 错误码 | 说明 |
|---|---|---|
400 |
| 请求参数格式错误或校验不通过 |
400 |
| 缺少必填参数 |
400 |
| 当前上下文不允许执行该请求 |
401 |
| 请求签名缺失或无效 |
403 |
| 身份校验失败 |
403 |
| 当前账号无权执行该查询 |
404 |
| 指定的库或资源不存在 |
429 |
| 超出配额或触发限流 |
500 |
| 服务内部错误 |
调用前,将 AK、SK 和 ResourceID 替换为真实值。
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="ListOpenVikingCollectionUser", body={ "ResourceID": "ov-20250609abcdef01", "Page": 1, "Limit": 20 } ) print("req http status code: ", http_code) print( "req result:\n", json.dumps(result_json, ensure_ascii=False, indent=2) )
如需按数据空间名称搜索,在 body 中增加 Keyword:
body={ "ResourceID": "ov-20250609abcdef01", "Keyword": "space", "Page": 1, "Limit": 20 }
以下为未传入 Keyword 时的示意响应,包含默认数据空间和一个自定义数据空间:
{ "ResponseMetadata": { "RequestId": "202609171030001A2B3C4D5E6F", "Action": "ListOpenVikingAccounts", "Version": "2025-06-09", "Service": "vikingdb", "Region": "cn-beijing" }, "Result": { "AccountList": [ { "OpenVikingAccountID": "default", "UserCount": 3, "CreateTime": "1732000000", "IsDefault": true }, { "OpenVikingAccountID": "space_a", "UserCount": 1, "CreateTime": "1732100000", "IsDefault": false } ], "Total": 2 } }