function test_chroma_collections
Maturity: 47
A diagnostic function that tests connectivity to ChromaDB instances across multiple connection methods and lists all available collections with their metadata.
File:
/tf/active/vicechatdev/test_chroma_collections.py
Lines:
18 - 131
18 - 131
Complexity:
moderate
moderate
Purpose
This function serves as a comprehensive diagnostic tool for ChromaDB deployments. It attempts to connect to ChromaDB using various host configurations (vice_chroma, localhost, 127.0.0.1), verifies connectivity through heartbeat checks, enumerates all collections, retrieves collection metadata and document counts, specifically searches for a '99_EDR' collection, tests collection access by peeking at documents, and falls back to testing a local persistent ChromaDB client. It's primarily used for troubleshooting connection issues and verifying database state.
Source Code
def test_chroma_collections():
"""Test Chroma DB connection and list all available collections."""
print("=" * 60)
print("CHROMA DB COLLECTIONS TEST")
print("=" * 60)
# Test different connection methods
connection_configs = [
{"host": "vice_chroma", "port": 8000, "name": "vice_chroma:8000"},
{"host": "localhost", "port": 8000, "name": "localhost:8000"},
{"host": "127.0.0.1", "port": 8000, "name": "127.0.0.1:8000"},
]
for config in connection_configs:
print(f"\n🔍 Testing connection to {config['name']}")
print("-" * 40)
try:
# Try to connect to Chroma DB
client = chromadb.HttpClient(host=config["host"], port=config["port"])
print(f"✅ Successfully connected to {config['name']}")
# Get heartbeat to verify connection
try:
heartbeat = client.heartbeat()
print(f"💓 Heartbeat: {heartbeat}")
except Exception as e:
print(f"⚠️ Heartbeat failed: {e}")
# List all collections
try:
collections = client.list_collections()
print(f"📁 Found {len(collections)} collections:")
if not collections:
print(" No collections found")
else:
for i, collection in enumerate(collections, 1):
print(f" {i}. {collection.name}")
# Get collection details
try:
count = collection.count()
print(f" - Document count: {count}")
# Try to get metadata
metadata = getattr(collection, 'metadata', None)
if metadata:
print(f" - Metadata: {metadata}")
except Exception as detail_error:
print(f" - Error getting details: {detail_error}")
# Specifically check for 99_EDR collection
print(f"\n🔍 Checking specifically for '99_EDR' collection...")
edr_found = False
for collection in collections:
if collection.name == "99_EDR":
edr_found = True
print(f"✅ Found 99_EDR collection!")
try:
count = collection.count()
print(f" - Document count: {count}")
except Exception as e:
print(f" - Error getting count: {e}")
break
if not edr_found:
print("❌ 99_EDR collection NOT found")
print(" Available collections:")
for collection in collections:
print(f" - {collection.name}")
# Try to get a collection (test access)
if collections:
test_collection = collections[0]
print(f"\n🧪 Testing access to '{test_collection.name}' collection...")
try:
# Try to query the collection
result = test_collection.peek(limit=1)
print(f"✅ Successfully accessed collection")
if result['ids']:
print(f" - Sample document ID: {result['ids'][0]}")
except Exception as access_error:
print(f"⚠️ Error accessing collection: {access_error}")
except Exception as list_error:
print(f"❌ Error listing collections: {list_error}")
except Exception as conn_error:
print(f"❌ Failed to connect to {config['name']}: {conn_error}")
# Test local persistent client as fallback
print(f"\n🔍 Testing local persistent client")
print("-" * 40)
try:
# Try local persistent client
local_client = chromadb.PersistentClient(path="./chroma_db")
print("✅ Successfully created local persistent client")
collections = local_client.list_collections()
print(f"📁 Local collections found: {len(collections)}")
for collection in collections:
print(f" - {collection.name}")
except Exception as local_error:
print(f"❌ Local persistent client failed: {local_error}")
print(f"\n" + "=" * 60)
print("COLLECTION TEST COMPLETED")
print("=" * 60)Return Value
This function does not return any value (implicitly returns None). It outputs diagnostic information directly to stdout using print statements, including connection status, collection names, document counts, metadata, and error messages.
Dependencies
chromadb
Required Imports
import chromadb
Usage Example
import chromadb
def test_chroma_collections():
# Function implementation here
pass
# Simply call the function to run diagnostics
test_chroma_collections()
# Expected output:
# ============================================================
# CHROMA DB COLLECTIONS TEST
# ============================================================
#
# 🔍 Testing connection to vice_chroma:8000
# ----------------------------------------
# ✅ Successfully connected to vice_chroma:8000
# 💓 Heartbeat: 1234567890
# 📁 Found 3 collections:
# 1. 99_EDR
# - Document count: 150
# ...Best Practices
- This function is intended for diagnostic and testing purposes only, not for production use
- Ensure ChromaDB server is running before executing this function to avoid connection errors
- The function tests multiple connection methods sequentially; the first successful connection provides the most reliable results
- Review the console output carefully to identify which connection method works in your environment
- The function includes error handling for each operation, so partial failures won't crash the entire diagnostic process
- If all remote connections fail, the local persistent client test provides a fallback verification method
- The '99_EDR' collection check is hardcoded; modify this section if testing for different specific collections
- Consider redirecting output to a log file for detailed analysis: python script.py > chroma_diagnostics.log 2>&1
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
function test_collection_creation 86.0% similar
-
function main_v16 75.0% similar
-
function quick_test 52.0% similar
-
function test_reference_system_completeness 50.0% similar
-
function check_filecloud_structure 49.8% similar