image_management_api/test_qdrant_connection.py
2025-05-24 20:13:11 +02:00

58 lines
1.8 KiB
Python

#!/usr/bin/env python3
"""
Simple test script to verify Qdrant connection
Auto-generated by deployment/scripts/setup_local_env.sh
"""
import os
import sys
from src.services.vector_db import VectorDatabaseService
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
def test_qdrant_connection():
"""Test the connection to Qdrant"""
# Set environment variables from deployed infrastructure
# os.environ['QDRANT_HOST'] = '34.171.134.17'
# os.environ['QDRANT_PORT'] = '6333'
# os.environ['QDRANT_HTTPS'] = 'false' # Explicitly disable HTTPS
# os.environ['QDRANT_PREFER_GRPC'] = 'false' # Explicitly disable gRPC
try:
print("Testing Qdrant connection...")
# print(f"Host: {os.environ['QDRANT_HOST']}")
# print(f"Port: {os.environ['QDRANT_PORT']}")
# print(f"HTTPS: {os.environ['QDRANT_HTTPS']}")
# print(f"gRPC: {os.environ['QDRANT_PREFER_GRPC']}")
# Initialize the service with explicit parameters to ensure HTTP is used
vector_db = VectorDatabaseService(
host=os.environ['QDRANT_HOST'],
port=int(os.environ['QDRANT_PORT']),
prefer_grpc=False,
https=False
)
# Test health check
is_healthy = vector_db.health_check()
print(f"Health check: {'✓ PASSED' if is_healthy else '✗ FAILED'}")
# Get collection info
collection_info = vector_db.get_collection_info()
print(f"Collection info: {collection_info}")
print("\n✓ Qdrant connection test PASSED!")
return True
except Exception as e:
print(f"\n✗ Qdrant connection test FAILED: {e}")
return False
if __name__ == "__main__":
success = test_qdrant_connection()
sys.exit(0 if success else 1)