35 lines
871 B
Python
35 lines
871 B
Python
|
import asyncio
|
||
|
|
|
||
|
|
from neo4j import GraphDatabase
|
||
|
|
from pymilvus import connections
|
||
|
|
from redis import Redis
|
||
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
|
|
||
|
|
from app.core.config import get_settings
|
||
|
|
|
||
|
|
|
||
|
|
async def check_mysql(dsn: str) -> None:
|
||
|
|
engine = create_async_engine(dsn)
|
||
|
|
async with engine.connect():
|
||
|
|
print("mysql=ok")
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
settings = get_settings()
|
||
|
|
asyncio.run(check_mysql(settings.mysql_dsn))
|
||
|
|
print(f"redis={Redis.from_url(settings.redis_url).ping()}")
|
||
|
|
driver = GraphDatabase.driver(
|
||
|
|
settings.neo4j_uri,
|
||
|
|
auth=(settings.neo4j_username, settings.neo4j_password),
|
||
|
|
)
|
||
|
|
driver.verify_connectivity()
|
||
|
|
print("neo4j=ok")
|
||
|
|
driver.close()
|
||
|
|
connections.connect(alias="default", uri=settings.milvus_uri)
|
||
|
|
print("milvus=ok")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|