Skip to main content

Python Client

The hlquery Python client provides a clean, Pythonic API for interacting with hlquery servers from Python applications.

Features​

  • No External Dependencies: Uses Python's built-in urllib and json modules
  • Pythonic API: Clean, intuitive interface
  • Type-safe Responses: Response objects with helper methods
  • Consistent API Design: Familiar structure for developers
  • Authentication Support: Bearer token and X-API-Key authentication
  • Comprehensive Validation: Input validation for all operations

Installation​

The Python client has no external dependencies. Install as a package:

cd api/python
pip install -e .

Or use directly:

import sys
sys.path.insert(0, 'api/python')
from lib import Client

Optional PDF Support​

If you want to index local PDF files directly through the Python client, install the optional parser:

pip install PyPDF2

The client exposes documents_api().add_pdf(collection, file_path, options=None) to parse the PDF locally, normalize the extracted content, and submit it as a regular hlquery document.

Quick Start​

Basic Usage​

from lib import Client

# Initialize client
client = Client('http://localhost:9200')

# Health check
health = client.health()
print(f"Status: {health.get_status_code()}")

# List collections
collections = client.list_collections(0, 10)
if collections.is_success():
    body = collections.get_body()
    print(f"Found {len(body.get('collections', []))} collections")

With Authentication​

# Method 1: Set token in constructor
client = Client('http://localhost:9200', {
    'token': 'your_token_here',
    'auth_method': 'bearer'  # or 'api-key'
})

# Method 2: Set token dynamically
client = Client('http://localhost:9200')
client.set_auth_token('your_token_here', 'bearer')

# Method 3: Use X-API-Key
client.set_auth_token('your_token_here', 'api-key')

API Reference​

Client Initialization​

client = Client(base_url, options=None)

Parameters:

  • base_url (str, required): Base URL of hlquery server (e.g., 'http://localhost:9200')
  • options (dict, optional): Client options
    • token (str): Authentication token
    • auth_method (str): Authentication method ('bearer' or 'api-key')
    • timeout (int): Request timeout in seconds

Health & Status​

health()​

Check server health status.

health = client.health()
if health.is_success():
    body = health.get_body()
    print(f"Status: {body.get('status')}")

stats()​

Get server statistics.

stats = client.stats()
body = stats.get_body()
print(f"Collections: {body.get('collections', {}).get('total')}")

info()​

Get server information.

info = client.info()

Collections API​

Using Collections API Object​

collections = client.collections_api()

# List collections
result = collections.list(0, 10)

# Get collection
result = collections.get('my_collection')

# Create collection
schema = {
    'fields': [
        {'name': 'title', 'type': 'string'},
        {'name': 'price', 'type': 'float'}
    ]
}
result = collections.create('new_collection', schema)

# Delete collection
result = collections.delete('collection_name')

# Update collection
result = collections.update('collection_name', {
    'fields': [{'name': 'new_field', 'type': 'string'}]
})

Direct Methods​

# List collections
collections = client.list_collections(offset, limit)

# Get collection
collection = client.get_collection('collection_name')

# Create collection
result = client.create_collection('collection_name', schema)

# Delete collection
result = client.delete_collection('collection_name')

Documents API​

Using Documents API Object​

documents = client.documents_api()

# List documents
result = documents.list('collection_name', 0, 10)

# Get document
result = documents.get('collection_name', 'document_id')

# Add document
doc = {
    'id': 'doc1',
    'title': 'Example',
    'price': 99.99
}
result = documents.add('collection_name', doc)

# Update document
result = documents.update('collection_name', 'doc_id', doc)

# Delete document
result = documents.delete('collection_name', 'doc_id')

# Import documents (bulk)
docs = [doc1, doc2, doc3]
result = documents.import_documents('collection_name', docs)

# Parse and add a local PDF file
pdf_result = documents.add_pdf('collection_name', './files/report.pdf', {
    'document': {
        'source_type': 'pdf'
    }
})

Direct Methods​

# List documents
docs = client.list_documents('collection_name', offset, limit)

# Get document
doc = client.get_document('collection_name', 'doc_id')

# Add document
result = client.add_document('collection_name', doc)

# Update document
result = client.update_document('collection_name', 'doc_id', doc)

# Delete document
result = client.delete_document('collection_name', 'doc_id')

# Import documents
result = client.import_documents('collection_name', docs)

Search API​

Using Search API Object​

search = client.search_api()

# Perform search
result = search.perform('collection_name', {
    'q': 'laptop',
    'query_by': 'title,description',
    'filter_by': 'price:>1000',
    'sort_by': 'price:asc',
    'limit': 10
})

# Multi-search
searches = [
    {'collection': 'products', 'q': 'laptop'},
    {'collection': 'articles', 'q': 'laptop'}
]
result = search.multi(searches)

# Vector search
result = search.vector('collection_name', {
    'vector': [0.1, 0.2, 0.3],
    'limit': 10
})

Direct Methods​

# Search
results = client.search('collection_name', {
    'q': 'laptop',
    'query_by': 'title'
})

# Multi-search
results = client.multi_search(searches)

# Vector search
results = client.vector_search('collection_name', {
    'vector': [0.1, 0.2, 0.3]
})

Response Objects​

All API methods return a Response object with helper methods:

response = client.health()

# Get HTTP status code
status_code = response.get_status_code()

# Get response body
body = response.get_body()

# Check if successful
if response.is_success():
    # Handle success
    pass

# Check if error
if response.is_error():
    error = response.get_error()
    print(f"Error: {error}")

# Convert to dict format (for backward compatibility)
data = response.to_dict()

Error Handling​

The client raises custom exceptions:

from lib.exceptions import (
    HlqueryException,
    AuthenticationException,
    RequestException
)

try:
    result = client.create_collection('collection', schema)
except AuthenticationException as e:
    print(f"Authentication failed: {e}")
except RequestException as e:
    print(f"Request failed: {e}")
except HlqueryException as e:
    print(f"Error: {e}")

Examples​

See the examples/ directory for complete examples:

  • basic_usage.py - Basic usage examples
  • collections.py - Collection management
  • documents.py - Document CRUD operations
  • search.py - Search operations

Testing​

The Python API includes a comprehensive test suite with 31 test scripts covering all API functionality.

Quick Start​

cd api/python/tests
python3 test_get_document.py
python3 test_search.py
python3 test_filter_search.py

Test Documentation​

All tests are self-contained, create their own test data, and include detailed developer comments explaining what each test does and how to run it.

Next Steps​