让我详细解析动态接口调用的实现。这个功能主要在 core/api.py 中实现，我们来看关键部分：
- 首先，看看公开接口的实现（private 接口实现类似，只是多了认证）：
@app_core.get(path="/public/{item_id}", description="公开接口", summary="该接口主要提供公开访问来获取数据")
def root(request: Request, item_id: str):
"""
接收请求参数及接口名称并返回 JSON 数据
:param request: 请求信息
:param item_id: 必选参数; 测试接口名 stock_dxsyl_em 来获取 打新收益率 数据
:return: 指定 接口名称 和 参数 的数据
"""
动态调用的核心实现包含以下几个关键步骤：
- 接口存在性验证：
interface_list = dir(ak) # 获取 akshare 模块中所有可用的接口
if item_id not in interface_list:
logger.info("未找到该接口，请升级 AKShare 到最新版本...")
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"error": "未找到该接口，请升级 AKShare..."
},
)
- 参数处理：
# URL 解码，处理 URL 中的特殊字符
decode_params = urllib.parse.unquote(str(request.query_params))
# 特殊处理 cookie 参数的情况
if "cookie" in decode_params:
eval_str = (
decode_params.split(sep="=", maxsplit=1)[0]
+ "='"
+ decode_params.split(sep="=", maxsplit=1)[1]
+ "'"
)
eval_str = eval_str.replace("+", " ")
else:
# 将 URL 参数转换为 Python 函数参数格式
eval_str = decode_params.replace("&", '", ').replace("=", '="') + '"'
eval_str = eval_str.replace("+", " ") # 处理空格
- 动态调用实现：
# 无参数调用
if not bool(request.query_params):
try:
# 使用 eval 动态执行 akshare 接口调用
received_df = eval("ak." + item_id + "()")
if received_df is None:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"error": "该接口返回数据为空，请确认参数是否正确"}
)
# 将返回的 DataFrame 转换为 JSON
temp_df = received_df.to_json(orient="records", date_format="iso")
except KeyError as e:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"error": f"请输入正确的参数错误 {e}..."
},
)
return JSONResponse(status_code=status.HTTP_200_OK, content=json.loads(temp_df))
else:
# 有参数调用
try:
# 构造带参数的函数调用
received_df = eval("ak." + item_id + f"({eval_str})")
if received_df is None:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"error": "该接口返回数据为空..."}
)
temp_df = received_df.to_json(orient="records", date_format="iso")
except KeyError as e:
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"error": f"请输入正确的参数错误 {e}..."
},
)
return JSONResponse(status_code=status.HTTP_200_OK, content=json.loads(temp_df))
让我用一个具体的例子来说明这个动态调用过程：
假设用户想要调用 AKShare 的 stock_zh_a_spot_em 接口获取 A 股实时行情数据，URL 可能是：
GET /api/public/stock_zh_a_spot_em
处理流程如下：
- 接口验证：
- 系统首先检查
stock_zh_a_spot_em是否在dir(ak)中存在 -
如果不存在，返回 404 错误
-
参数处理：
- 如果 URL 带有参数，比如：
/api/public/stock_zh_a_spot_em?symbol=000001 decode_params会得到symbol=000001-
转换后的
eval_str会变成symbol="000001" -
动态执行：
- 系统会构造并执行：
eval("ak.stock_zh_a_spot_em(symbol=\"000001\")") -
这相当于在 Python 中直接调用：
ak.stock_zh_a_spot_em(symbol="000001") -
结果处理：
- 获取到的 DataFrame 数据会被转换为 JSON 格式
- 通过 HTTP 响应返回给客户端
这种实现的优点： 1. 灵活性：可以动态调用 AKShare 中的任何接口，无需为每个接口单独编写代码 2. 可扩展性：当 AKShare 添加新接口时，无需修改 AKTools 代码 3. 参数兼容：支持各种参数类型的传递，包括特殊情况（如 cookie）
需要注意的安全考虑：
1. 使用 eval 进行动态调用可能存在安全风险，但这里通过以下方式降低风险：
- 严格验证接口名称必须存在于 dir(ak) 中
- 参数值都被作为字符串处理
- 只允许调用 ak 模块中的方法
这就是 AKTools 动态接口调用的核心实现原理。它通过 Python 的动态特性，巧妙地将 HTTP 请求转换为 AKShare 接口调用，实现了一个灵活且可扩展的 API 系统。