Python SDK

The official Torque Python SDK provides an HTTP client and TCP binary ingest client. Zero external dependencies — uses only the Python standard library. Requires Python 3.10+.

Installation

pip install torque-http

HTTP Client

The TorqueHttpClient class provides full access to the Torque REST API:

from torque_http import TorqueHttpClient

client = TorqueHttpClient("http://localhost:8108", api_key="YOUR_API_KEY")

# Health check
health = client.health()

# List collections
collections = client.list_collections()

# Search
results = client.search("products", q="running shoes", query_by="title,description",
                         filter_by="in_stock:true", sort_by="price:asc", per_page=20)

Multi-Node with Failover

client = TorqueHttpClient(
    ["http://node1:8108", "http://node2:8108"],
    api_key="YOUR_API_KEY",
    num_retries=5,
)

Typed Search Parameters

from torque_http import TorqueHttpClient, SearchParameters

client = TorqueHttpClient("http://localhost:8108", api_key="YOUR_API_KEY")

params = SearchParameters(q="laptop", query_by="title", per_page=20)
results = client.search("products", params)

for hit in results.hits:
    print(f"{hit.document['title']} - {hit.document['price']}")

Grouped & Multi-Search

# Grouped search
grouped = client.search_grouped("products", q="shoes", query_by="title",
                                 group_by="brand", group_limit=3)

for group in grouped.grouped_hits:
    print(f"Brand: {group.group_key} ({group.found} hits)")

# Multi-search
results = client.multi_search([
    {"collection": "products", "q": "shoes", "query_by": "title"},
    {"collection": "articles", "q": "shoes", "query_by": "body"},
])

Collection Management

from torque_http import CollectionSchema, FieldDefinition

# Create a collection
client.create_collection(CollectionSchema(
    name="products",
    fields=[
        FieldDefinition(name="title", type="string"),
        FieldDefinition(name="price", type="float", sort=True),
        FieldDefinition(name="category", type="string", facet=True),
    ],
    default_sorting_field="price",
))

# Delete a collection
client.delete_collection("products")

Document Operations

# Import documents (JSONL)
results = client.import_documents("products", [
    {"id": "1", "title": "Running Shoes", "price": 89.99, "category": "Footwear"},
    {"id": "2", "title": "Laptop Stand", "price": 49.99, "category": "Electronics"},
], action="upsert")

# Export documents
for doc in client.export_documents("products"):
    print(doc)

# CRUD
client.create_document("products", {"id": "3", "title": "Mouse", "price": 29.99})
doc = client.get_document("products", "3")
client.update_document("products", "3", {"price": 24.99})
client.delete_document("products", "3")

Aliases, Synonyms, Presets & More

# Aliases
client.upsert_alias("products", "products_v2")
aliases = client.list_aliases()

# API keys
key = client.create_key("Search only", ["documents:search"], ["products"])

# Synonyms, presets, stopwords
client.upsert_synonym_set("my_synonyms", [{"root": "laptop", "synonyms": ["notebook", "computer"]}])
client.upsert_preset("my_preset", {"value": {"q": "default", "query_by": "title"}})
client.upsert_stopwords("my_stopwords", ["the", "a", "an", "is"])

TCP Binary Ingestion

The TorqueIngestClient provides high-throughput document streaming via the binary TCP protocol:

from torque_http import TorqueIngestClient, DocumentEncoder

# Schema dict for binary encoding
schema = {"fields": [
    {"name": "title", "type": "string"},
    {"name": "price", "type": "float"},
    {"name": "category", "type": "string"},
]}
encoder = DocumentEncoder(schema)

with TorqueIngestClient() as ingest:
    ingest.connect("localhost", 8109)
    ingest.start_ingest("products")  # mode=0 (batch) is default

    # Send documents in batches — encoder handles binary encoding
    docs = []
    for i in range(100000):
        docs.append({"id": f"product_{i}", "title": f"Product {i}",
                      "price": 9.99 + i * 0.01, "category": "Electronics"})
        if len(docs) == 5000:
            ingest.send_batch(docs, encoder)
            docs.clear()

    if docs:
        ingest.send_batch(docs, encoder)

    result = ingest.commit()
    print(f"Committed batch: seq={result.batch_seq}, docs={result.docs_received}")

TQBF Binary File Writer

The TorqueBinaryFileWriter creates TQBF binary files for bulk import via POST /collections/{name}/documents/import-binary:

from torque_http import TorqueBinaryFileWriter, DocumentEncoder

schema = {"fields": [
    {"name": "title", "type": "string"},
    {"name": "price", "type": "float"},
]}
encoder = DocumentEncoder(schema)

with TorqueBinaryFileWriter("output.tqbf", encoder) as writer:
    for i in range(100000):
        writer.write({"id": f"p_{i}", "title": f"Product {i}", "price": 9.99})

# Upload via HTTP
# curl -X POST -H "X-TYPESENSE-API-KEY: KEY" \
#   "http://localhost:8108/collections/products/documents/import-binary" \
#   --data-binary @output.tqbf

HMAC Scoped Keys

scoped_key = client.generate_scoped_search_key(
    "your-search-key",
    {"filter_by": "company_id:=123"},
)

Error Handling

HTTP API errors raise typed exceptions:

TCP protocol errors raise TorqueProtocolError with an error code, and connection issues raise TorqueConnectionError.

Models

The SDK includes typed dataclasses for all API responses:

• SearchResult, GroupedSearchResult, Hit — search responses
• CollectionSchema, FieldDefinition, CollectionResponse — collection schemas
• ApiKey, CollectionAlias, SynonymSet, Preset, StopwordSet — management resources
• ImportResult, DeleteResult, UpdateResult — mutation results
• AdminStats, HealthResponse — server status