54 lines
1.7 KiB
Plaintext
54 lines
1.7 KiB
Plaintext
def _escape(value):
|
||
if value is None:
|
||
return None
|
||
return str(value).replace("'", "''")
|
||
|
||
|
||
async def get_model_api_doc(ns={}):
|
||
"""
|
||
根据 model_id 查询模型 API 文档。
|
||
|
||
参数：
|
||
model_id (str) 模型ID，必填
|
||
|
||
返回 data 字段：
|
||
id, model_id, curl_code, python_code, created_at, updated_at
|
||
"""
|
||
model_id = ns.get('id')
|
||
if not model_id:
|
||
return {'status': False, 'msg': 'model id is required'}
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext('kboss') as sor:
|
||
try:
|
||
# 通过model_id从model_management表中查询model_name
|
||
model_name_sql = """
|
||
SELECT model_name FROM model_management WHERE id = '%s' LIMIT 1;
|
||
""" % _escape(model_id)
|
||
model_name = await sor.sqlExe(model_name_sql, {})
|
||
if not model_name:
|
||
return {'status': False, 'msg': 'model not found'}
|
||
model_name = model_name[0]['model_name']
|
||
|
||
find_sql = """
|
||
SELECT id, api_url, model_id, curl_code, python_code, created_at, updated_at
|
||
FROM model_api_doc
|
||
WHERE model_id = '%s'
|
||
LIMIT 1;
|
||
""" % _escape(model_id)
|
||
result = await sor.sqlExe(find_sql, {})
|
||
if not result:
|
||
return {'status': False, 'msg': 'api doc not found'}
|
||
result[0]['model_name'] = model_name
|
||
return {
|
||
'status': True,
|
||
'msg': 'get model api doc success',
|
||
'data': result[0],
|
||
}
|
||
except Exception as e:
|
||
return {'status': False, 'msg': 'get model api doc failed, %s' % str(e)}
|
||
|
||
|
||
ret = await get_model_api_doc(params_kw)
|
||
return ret
|