TenderAgent/TenderAgent完整代码/check_env.py

import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from pymilvus import connections, utility

DB_URL = "postgresql+asyncpg://tenderagent_user:123456@localhost:5433/tenderagent"

async def check_postgres():
    engine = create_async_engine(DB_URL)
    async with engine.connect() as conn:
        r = await conn.execute(text("SELECT 1"))
        print("✅ PostgreSQL 连通：", r.scalar())
    await engine.dispose()

def check_milvus():
    connections.connect(alias="default", host="localhost", port="19531")
    print("✅ Milvus 连通，已有集合：", utility.list_collections())

async def main():
    await check_postgres()
    check_milvus()

asyncio.run(main())