41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""查询表现有数据的结构"""
|
|
import asyncio
|
|
import httpx
|
|
import json
|
|
|
|
COZE_API_BASE = "https://api.coze.cn"
|
|
COZE_PAT_TOKEN = "pat_nd1wU47WyPS9GCIyJ1clnH8h1WOQXGrYELX8w73TnSZaYbFdYD4swIhzcETBUbfT"
|
|
WORKFLOW_ID = "7597376294612107318"
|
|
|
|
async def query(table: str, sql: str):
|
|
headers = {
|
|
"Authorization": f"Bearer {COZE_PAT_TOKEN}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
input_json = json.dumps({"table": table, "sql": sql}, ensure_ascii=False)
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(
|
|
f"{COZE_API_BASE}/v1/workflow/run",
|
|
headers=headers,
|
|
json={"workflow_id": WORKFLOW_ID, "parameters": {"input": input_json}}
|
|
)
|
|
result = response.json()
|
|
print(f"\n{table}:")
|
|
print(f" code: {result.get('code')}")
|
|
if result.get('data'):
|
|
data = json.loads(result.get('data', '{}'))
|
|
output = data.get('output', [])
|
|
if output and len(output) > 0:
|
|
print(f" 列名: {list(output[0].keys())}")
|
|
else:
|
|
print(f" 空数据")
|
|
|
|
async def main():
|
|
print("查询表结构...")
|
|
await query("assessments", "SELECT * FROM ci_interview_assessments LIMIT 1")
|
|
await query("logs", "SELECT * FROM ci_interview_logs LIMIT 1")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|