| 1234567891011121314151617181920212223242526272829303132 |
- from __future__ import annotations
- import time
- from pymilvus import MilvusClient
- from app.config import get_settings
- def main() -> None:
- settings = get_settings()
- kwargs = {"uri": settings.milvus_uri}
- token = settings.milvus_token.get_secret_value()
- if token:
- kwargs["token"] = token
- deadline = time.monotonic() + 120
- last_error: Exception | None = None
- while time.monotonic() < deadline:
- try:
- client = MilvusClient(**kwargs)
- client.list_collections()
- print(f"Milvus 已就绪：{settings.milvus_uri}")
- return
- except Exception as exc:
- last_error = exc
- time.sleep(2)
- raise SystemExit(f"Milvus 在 120 秒内未就绪：{last_error}")
- if __name__ == "__main__":
- main()
|
