(session: SessionDep, mcp_ds: McpDs)
| 106 | |
| 107 | @router.post("/mcp_ds_list", operation_id="mcp_datasource_list") |
| 108 | async def datasource_list(session: SessionDep, mcp_ds: McpDs): |
| 109 | session_user = get_user(session, mcp_ds.token) |
| 110 | if mcp_ds.oid: |
| 111 | session_user.oid = int(mcp_ds.oid) |
| 112 | ds_list = get_datasource_list(session=session, user=session_user) |
| 113 | result = [] |
| 114 | for item in ds_list: |
| 115 | dic = item.__dict__ |
| 116 | dic.pop('embedding', None) |
| 117 | dic.pop('table_relation', None) |
| 118 | dic.pop('recommended_config', None) |
| 119 | dic.pop('configuration', None) |
| 120 | result.append(dic) |
| 121 | return result |
| 122 | |
| 123 | |
| 124 | # |
nothing calls this directly
no test coverage detected