38 lines
1023 B
Python
38 lines
1023 B
Python
import asyncio
|
|
import redis.asyncio as redis
|
|
from neo4j import AsyncGraphDatabase
|
|
from pymilvus import connections
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
async def main() -> None:
|
|
settings = get_settings()
|
|
cache = redis.from_url(settings.redis_url)
|
|
try:
|
|
print("redis", await cache.ping())
|
|
except Exception as exc:
|
|
print("redis", f"unavailable:{type(exc).__name__}")
|
|
finally:
|
|
await cache.aclose()
|
|
driver = AsyncGraphDatabase.driver(
|
|
settings.neo4j_uri,
|
|
auth=(settings.neo4j_username, settings.neo4j_password),
|
|
)
|
|
try:
|
|
await driver.verify_connectivity()
|
|
print("neo4j", "ok")
|
|
finally:
|
|
await driver.close()
|
|
try:
|
|
connections.connect(alias="default", uri=settings.resolved_milvus_uri)
|
|
print("milvus", "ok")
|
|
except Exception as exc:
|
|
print("milvus", f"unavailable:{type(exc).__name__}")
|
|
finally:
|
|
connections.disconnect("default")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|