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