Get the status of a transaction
(client, coll_name, txn_id)
| 2 | import time |
| 3 | |
| 4 | def get_transaction_status(client, coll_name, txn_id): |
| 5 | """Get the status of a transaction""" |
| 6 | host = client.host |
| 7 | url = f"{host}/vectordb/collections/{coll_name}/transactions/{txn_id}/status" |
| 8 | resp = requests.get(url, headers=client._get_headers(), verify=False) |
| 9 | result = resp.json() |
| 10 | return result['status'] |
| 11 | |
| 12 | def poll_transaction_completion(client, collection_name, txn_id, target_status='complete', |
| 13 | max_attempts=10, sleep_interval=1): |
no test coverage detected