AsyncTest

Appearance

全局数据源 ​

什么是全局数据源？ ​

全局数据源是 AsyncTest 在自定义脚本中提供的 全局数据访问对象，用于在脚本中查询和操作用例关联的全局数据源数据。

系统会将分组的原始数据自动扁平化为一个完整的列表，你可以通过列查询、行查询、交叉匹配等方式灵活获取数据。

你可以通过 at.get_source 直接获取全局数据源对象，无需初始化。

基本用法 ​

python
# 获取全局数据源对象
source = at.get_source

# 查看所有数据
all_data = source.all_data
print(all_data)

数据结构 ​

全局数据源会将多组数据扁平化为一个列表，每个元素是一个字典（一行数据）：

python
# 原始分组数据会被扁平化为：
[
    {"name": "Sheldon", "$ast_set_name": "数据_hu7Xe", "age": "18"},
    {"name": "Tom", "$ast_set_name": "数据_QtZIW", "age": "29"},
    {"private_name": "Jone", "$ast_set_name": "数据_HXU2d", "private_age": "18岁"},
    {"private_name": "Yuki", "$ast_set_name": "数据_XvkQF", "private_age": "100岁"},
    ...
]

不同组的数据可能拥有不同的字段，查询时会自动跳过不包含目标字段的行。

属性 ​

all_data ​

获取完整的扁平化数据列表

python
source = at.get_source
all_data = source.all_data

返回值： 列表，包含所有行数据的字典

示例：

python
source = at.get_source
all_data = source.all_data
print(f"共 {len(all_data)} 条数据")
for row in all_data:
    print(row)

方法列表 ​

get_column() ​

获取某一列的所有值

python
source.get_column(column_name)

参数：

  • column_name: 列名（字符串）

返回值： 列表，包含该列所有值（自动跳过不包含该字段的行）

示例：

python
source = at.get_source

# 获取 name 列的所有值
names = source.get_column("name")
print(names)  # ['Sheldon', 'Tom', 'Sheldon', 'Tom', 'Yummy']

# 获取 age 列的所有值（不包含 age 字段的行会被跳过）
ages = source.get_column("age")
print(ages)  # ['18', '29']

query_rows() ​

根据字段值查询行数据

python
source.query_rows(key, value, match_type='equals', return_index=None)

参数：

  • key: 要匹配的字段名
  • value: 要匹配的值
  • match_type: 匹配方式（默认 'equals'）
    • 'equals'：精确匹配（会将两边转为字符串比较）
    • 'contains'：模糊匹配（包含关系）
  • return_index: 返回结果列表中的第 n 个元素（默认 None 返回全部）

返回值：

  • return_index=None 时：返回匹配的行列表
  • return_index=n 时：返回第 n 个匹配结果（字典），下标越界返回 None

示例：

精确匹配 ​

python
source = at.get_source

# 查找 name 为 Sheldon 的所有行
results = source.query_rows(key="name", value="Sheldon")
print(f"找到 {len(results)} 条数据")
# [
#     {"name": "Sheldon", "$ast_set_name": "数据_hu7Xe", "age": "18"},
#     {"name": "Sheldon", "$ast_set_name": "数据_SA7Zn", "index": "1"}
# ]

模糊匹配 ​

python
source = at.get_source

# 查找 index 包含 '99' 的行
results = source.query_rows(key="index", value="99", match_type="contains")
print(results)
# [{"name": "Tom", "$ast_set_name": "数据_BvSJq", "index": "1990"}]

指定返回下标 ​

python
source = at.get_source

# 查找 name 为 Sheldon 的第 1 个结果（下标从 0 开始）
first = source.query_rows(key="name", value="Sheldon", return_index=0)
print(first)
# {"name": "Sheldon", "$ast_set_name": "数据_hu7Xe", "age": "18"}

# 查找 name 为 Sheldon 的第 2 个结果
second = source.query_rows(key="name", value="Sheldon", return_index=1)
print(second)
# {"name": "Sheldon", "$ast_set_name": "数据_SA7Zn", "index": "1"}

# 下标越界返回 None
result = source.query_rows(key="name", value="Sheldon", return_index=99)
print(result)  # None

query_by_intersection() ​

通过外部列表与全局数据源进行交叉匹配

python
source.query_by_intersection(external_list, external_key, data_key)

参数：

  • external_list: 外部数据列表（字典列表）
  • external_key: 外部列表中用于匹配的字段名
  • data_key: 全局数据源中用于匹配的字段名

返回值： 列表，包含匹配到的全局数据源行

逻辑： 只要外部列表某元素的 external_key 值出现在全局数据源的 data_key 中，就返回全局数据源的该行。

示例：

python
source = at.get_source

# 外部列表（例如从接口响应中获取的数据）
external_list = [
    {"assigneeName": "Yummy", "task": "A"},
    {"assigneeName": "Jone", "task": "B"},
    {"assigneeName": "Nobody", "task": "C"}
]

# 交叉匹配：外部列表的 assigneeName 匹配全局数据源的 name
results = source.query_by_intersection(
    external_list=external_list,
    external_key="assigneeName",
    data_key="name"
)
print(results)
# 返回 name="Yummy" 的行（Jone 在 private_name 字段中，不会匹配 name 字段）

完整使用示例 ​

根据数据源动态设置变量 ​

python
source = at.get_source

# 获取第一条数据的 name 和 age
first_row = source.query_rows(key="name", value="Sheldon", return_index=0)
if first_row:
    at.temp.set("username", first_row.get("name"))
    at.temp.set("user_age", first_row.get("age"))

遍历数据源进行批量操作 ​

python
source = at.get_source

# 获取所有 name
names = source.get_column("name")
print(f"共 {len(names)} 个用户名: {names}")

# 遍历所有数据
for row in source.all_data:
    print(row)

结合接口响应进行交叉匹配 ​

python
source = at.get_source

# 从接口响应中获取用户列表
response_data = await at.response.async_json()
user_list = response_data.get("data", {}).get("users", [])

# 交叉匹配：找出接口返回的用户中，在全局数据源里存在的数据
matched = source.query_by_intersection(
    external_list=user_list,
    external_key="username",
    data_key="name"
)
print(f"匹配到 {len(matched)} 条数据")

# 将匹配结果保存到临时变量
import json
at.temp.set("matched_users", json.dumps(matched, ensure_ascii=False))

注意事项 ​

  • 通过 at.get_source 直接获取，无需初始化
  • 全局数据源会自动将分组数据扁平化为一个完整列表
  • 不同组的数据可能拥有不同的字段，查询时会自动跳过不包含目标字段的行
  • query_rows() 在比较时会将两边转为字符串，避免类型不一致导致匹配失败
  • query_rows() 的 return_index 下标从 0 开始，越界返回 None

总结 ​

全局数据源提供了灵活的数据查询能力：

  • 通过 at.get_source 直接访问
  • 支持列查询、行查询（精确/模糊）、交叉匹配等多种查询方式
  • 自动扁平化分组数据，统一访问
  • 适合在脚本中动态获取测试数据、进行数据驱动测试