LQDataGovernance/test/test_pymilvus.py

from pymilvus import connections, utility
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

def get_connection_info():
    """获取 Milvus 连接信息"""
    """连接到多个数据库"""
    databases = ['default', 'lq_db']
    try:
        # 使用与你的代码相同的连接方式  alias="default",  db_name="lq_db"
        connections.connect(
            alias="default", 
            host='192.168.0.3', 
            port='19530',
            db_name="lq_db"
        )
        
        #utility.get_connection("lq_db")
        # 获取服务器信息
        version = utility.get_server_version()
        print(f"✅ Milvus 版本: {version}")
        
        # 获取连接详情
        # conn = connections.get_connection("default")
        # print(f"✅ 连接地址: {conn._address}")
        # print(f"✅ 连接端口: {conn._port}")
        
        # 测试列出 collections
        collections = utility.list_collections()
        print(f"✅ 现有 collections: {collections}")
        print(f"{collections[0]}")

        collection_name = "tv_basis_of_preparation"
        collection = Collection(collection_name, using="lq_db")
        print(f"✅ 集合信息: {collection.name}")

        
        return True
        
    except Exception as e:
        print(f"❌ 连接失败: {e}")
        return False

if __name__ == "__main__":
    get_connection_info()