Post data to OpenSearchDB. :param json_data: Data to post, type dict or list. :param project: Name of the project. :return: bool, True if post successful, False otherwise.
(json_data, project)
| 168 | |
| 169 | @staticmethod |
| 170 | def postToOpenSearchDB(json_data, project) -> bool: |
| 171 | """ |
| 172 | Post data to OpenSearchDB. |
| 173 | |
| 174 | :param json_data: Data to post, type dict or list. |
| 175 | :param project: Name of the project. |
| 176 | :return: bool, True if post successful, False otherwise. |
| 177 | """ |
| 178 | use_poc_db = "sandbox" in project |
| 179 | if not OPEN_SEARCH_DB_BASE_URL: |
| 180 | OpenSearchDB.logger.info("OPEN_SEARCH_DB_BASE_URL is not set") |
| 181 | return False |
| 182 | if not use_poc_db and (not OPEN_SEARCH_DB_USERNAME |
| 183 | or not OPEN_SEARCH_DB_PASSWORD): |
| 184 | OpenSearchDB.logger.info( |
| 185 | "OPEN_SEARCH_DB_USERNAME or OPEN_SEARCH_DB_PASSWORD is not set") |
| 186 | return False |
| 187 | if not use_poc_db and project not in WRITE_ACCESS_PROJECT_NAME: |
| 188 | OpenSearchDB.logger.info( |
| 189 | f"project {project} is not in write access project list: {json.dumps(WRITE_ACCESS_PROJECT_NAME)}" |
| 190 | ) |
| 191 | return False |
| 192 | if not OpenSearchDB.typeCheckForOpenSearchDB(json_data): |
| 193 | OpenSearchDB.logger.info( |
| 194 | f"OpenSearchDB type check failed! json_data:{json_data}") |
| 195 | return False |
| 196 | |
| 197 | json_data_dump = json.dumps(json_data) |
| 198 | |
| 199 | if DISABLE_OPEN_SEARCH_DB_FOR_LOCAL_TEST: |
| 200 | OpenSearchDB.logger.info( |
| 201 | f"OpenSearchDB is disabled for local test, skip posting to OpenSearchDB: {json_data_dump}" |
| 202 | ) |
| 203 | return True |
| 204 | |
| 205 | url = f"{OPEN_SEARCH_DB_BASE_URL}/dataflow2/{project}/posting" |
| 206 | headers = { |
| 207 | "Content-Type": "application/json", |
| 208 | "Accept-Charset": "UTF-8" |
| 209 | } |
| 210 | |
| 211 | for attempt in range(DEFAULT_RETRY_COUNT): |
| 212 | try: |
| 213 | args = { |
| 214 | "url": url, |
| 215 | "data": json_data_dump, |
| 216 | "headers": headers, |
| 217 | "timeout": POST_TIMEOUT_SECONDS, |
| 218 | } |
| 219 | if not use_poc_db: |
| 220 | args["auth"] = HTTPProxyAuth(OPEN_SEARCH_DB_USERNAME, |
| 221 | OPEN_SEARCH_DB_PASSWORD) |
| 222 | res = requests.post(**args) |
| 223 | if res.status_code in (200, 201, 202): |
| 224 | if res.status_code != 200 and project == JOB_PROJECT_NAME: |
| 225 | OpenSearchDB.logger.info( |
| 226 | f"OpenSearchDB post not 200, log:{res.status_code} {res.text}" |
| 227 | ) |
no test coverage detected