| 107 | |
| 108 | |
| 109 | class AssistantOutDs: |
| 110 | assistant: AssistantHeader |
| 111 | ds_list: Optional[list[AssistantOutDsSchema]] = None |
| 112 | certificate: Optional[str] = None |
| 113 | request_origin: Optional[str] = None |
| 114 | |
| 115 | def __init__(self, assistant: AssistantHeader): |
| 116 | self.assistant = assistant |
| 117 | self.ds_list = None |
| 118 | self.certificate = assistant.certificate |
| 119 | self.request_origin = assistant.request_origin |
| 120 | self.get_ds_from_api() |
| 121 | |
| 122 | # @cache(namespace=CacheNamespace.EMBEDDED_INFO, cacheName=CacheName.ASSISTANT_DS, keyExpression="current_user.id") |
| 123 | def get_ds_from_api(self): |
| 124 | config: dict[any] = json.loads(self.assistant.configuration) |
| 125 | endpoint: str = config['endpoint'] |
| 126 | endpoint = self.get_complete_endpoint(endpoint=endpoint) |
| 127 | if not endpoint: |
| 128 | raise Exception( |
| 129 | f"Failed to get datasource list from {config['endpoint']}, error: [Assistant domain or endpoint miss]") |
| 130 | certificateList: list[any] = json.loads(self.certificate) |
| 131 | header = {} |
| 132 | cookies = {} |
| 133 | param = {} |
| 134 | for item in certificateList: |
| 135 | if item['target'] == 'header': |
| 136 | header[item['key']] = item['value'] |
| 137 | if item['target'] == 'cookie': |
| 138 | cookies[item['key']] = item['value'] |
| 139 | if item['target'] == 'param': |
| 140 | param[item['key']] = item['value'] |
| 141 | timeout = int(config.get('timeout')) if config.get('timeout') else 10 |
| 142 | res = requests.get(url=endpoint, params=param, headers=header, cookies=cookies, timeout=timeout) |
| 143 | if res.status_code == 200: |
| 144 | result_json: dict[any] = json.loads(res.text) |
| 145 | if result_json.get('code') == 0 or result_json.get('code') == 200: |
| 146 | temp_list = result_json.get('data', []) |
| 147 | temp_ds_list = [ |
| 148 | self.convert2schema(item, config) |
| 149 | for item in temp_list |
| 150 | ] |
| 151 | self.ds_list = temp_ds_list |
| 152 | return self.ds_list |
| 153 | else: |
| 154 | raise Exception(f"Failed to get datasource list from {endpoint}, error: {result_json.get('message')}") |
| 155 | else: |
| 156 | SQLBotLogUtil.error(f"Failed to get datasource list from {endpoint}, response: {res}") |
| 157 | raise Exception(f"Failed to get datasource list from {endpoint}, response: {res}") |
| 158 | |
| 159 | def get_first_element(self, text: str): |
| 160 | parts = re.split(r'[,;]', text.strip()) |
| 161 | first_domain = parts[0].strip() |
| 162 | return first_domain |
| 163 | |
| 164 | def get_complete_endpoint(self, endpoint: str) -> str | None: |
| 165 | if endpoint.startswith("http://") or endpoint.startswith("https://"): |
| 166 | return endpoint |