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 debugging 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). All output is printed to stdout, 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
# 2. my_collection
# - Document count: 42
# ...Best Practices
- Run this function in a development or testing environment to verify ChromaDB setup before production use
- Ensure ChromaDB server is running before executing this function to avoid connection errors
- Review the printed output to identify which connection method works in your environment
- Use this function to verify that expected collections exist and contain the anticipated number of documents
- The function tests multiple connection methods sequentially; the first successful connection indicates the correct configuration
- If all remote connections fail, the function falls back to testing a local persistent client
- This function is intended for diagnostic purposes only and should not be used in production code paths
- The './chroma_db' path for local persistent client may need to be adjusted based on your directory structure
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
function test_collection_creation 86.0% similar
-
function main_v31 74.9% similar
-
function load_data_from_chromadb_v1 65.2% similar
-
function reset_collection 65.2% similar
-
function main_v58 63.7% similar