worqhat 4.2.0

Python SDK for Worqhat API

Worqhat Python SDK

Official Python SDK for the Worqhat API. This SDK provides a type-safe interface to interact with Worqhat's database, workflows, and storage services.

Installation

pip install worqhat

Quick Start

from worqhat import Worqhat

client = Worqhat(api_key="your-api-key")
# Or use environment variable: export WORQHAT_API_KEY=your-api-key
# client = Worqhat()

response = client.db.execute_query(query="SELECT * FROM users")
print(response["data"])

Configuration

Basic Configuration

from worqhat import Worqhat

client = Worqhat(api_key="your-api-key")

Advanced Configuration

client = Worqhat(
    api_key="your-api-key",
    base_url="https://api.worqhat.app",  # Custom base URL (optional)
    target="app",  # Backend target: 'app' or 'fyi' (optional)
    timeout=60,  # Request timeout in seconds (optional)
)

Using Environment Variables

import os

# Set environment variable
os.environ["WORQHAT_API_KEY"] = "your-api-key"

# API key will be read automatically
client = Worqhat()

Context Manager

with Worqhat(api_key="your-api-key") as client:
    response = client.db.execute_query(query="SELECT * FROM users")
    print(response["data"])
# Session is automatically closed

Database Operations

Execute Raw SQL Query

Execute SQL queries with named or positional parameters.

Named Parameters

response = client.db.execute_query(
    query="SELECT * FROM users WHERE email = {email} AND status = {status}",
    params={"email": "[e-mail]", "status": "active"},
    environment="production"  # optional
)

print(response["data"])

Positional Parameters

response = client.db.execute_query(
    query="SELECT * FROM users WHERE email = $1 AND status = $2",
    params=["[e-mail]", "active"]
)

print(response["data"])

Insert Data

Insert single or multiple records into a table.

Single Record

response = client.db.insert(
    table="users",
    data={
        "email": "[e-mail]",
        "name": "New User",
        "status": "active"
    }
)

print(response["data"])

Multiple Records

response = client.db.insert(
    table="users",
    data=[
        {"email": "[e-mail]", "name": "User 1"},
        {"email": "[e-mail]", "name": "User 2"}
    ]
)

print(f"Inserted {response['count']} records")

Update Data

response = client.db.update(
    table="users",
    data={"status": "inactive"},
    where={"email": "[e-mail]"}
)

print(f"Updated {response['count']} records")

Delete Data

response = client.db.delete(
    table="users",
    where={"id": "123"}
)

print(response["message"])

List Tables

response = client.db.list_tables(environment="production", schema="public")

for table in response["tables"]:
    print(table["table_name"])

Get Table Schema

response = client.db.get_table_schema("users", environment="production")

print("Columns:", response["schema"]["columns"])
print("Indexes:", response["schema"]["indexes"])
print("Constraints:", response["schema"]["constraints"])

Get Table Count

# Get total count
response = client.db.get_table_count("users")
print(f"Total users: {response['count']}")

# Get count with filters
response = client.db.get_table_count("users", filters={"status": "active"})
print(f"Active users: {response['count']}")

Batch Operations

Execute multiple operations in a single transaction.

response = client.db.batch(
    operations=[
        {
            "type": "insert",
            "table": "users",
            "data": {"email": "[e-mail]", "name": "User 1"}
        },
        {
            "type": "update",
            "table": "users",
            "data": {"status": "active"},
            "where": {"email": "[e-mail]"}
        },
        {
            "type": "query",
            "query": "SELECT * FROM users WHERE email = $1",
            "params": ["[e-mail]"]
        }
    ],
    transactional=True  # All operations succeed or all fail
)

print(response["results"])

Vector Search Operations

Semantic Search

Perform semantic similarity search using vector embeddings.

Single Table Search

response = client.db.semantic_search(
    query="find products similar to wireless headphones",
    table="products",
    limit=10,
    threshold=0.7,
    filters={"category": "electronics"}
)

for result in response["results"]:
    print(f"{result['record']['name']} - Similarity: {result['similarity']}")

Cross-Table Search

response = client.db.semantic_search(
    query="customer support issues",
    tables=["tickets", "support_requests", "feedback"],
    limit=20,
    threshold=0.6
)

for result in response["results"]:
    print(f"Table: {result['table']}, Similarity: {result['similarity']}")

Hybrid Search

Combine semantic similarity with keyword search for better relevance.

response = client.db.hybrid_search(
    query="affordable wireless headphones",
    table="products",
    text_columns=["name", "description", "tags"],
    semantic_weight=0.7,
    keyword_weight=0.3,
    limit=10
)

for result in response["results"]:
    print(f"Combined Score: {result['combined_score']}")
    print(f"Semantic: {result['semantic_score']}, Keyword: {result['keyword_score']}")

Find Similar Records

Find records similar to a specific record.

response = client.db.find_similar(
    table="products",
    record_id="123",
    limit=5,
    threshold=0.75,
    exclude_self=True
)

print("Source Record:", response["source_record"])
for similar in response["similar_records"]:
    print(f"{similar['record']['name']} - Similarity: {similar['similarity']}")

Workflow Operations

Trigger a Workflow

response = client.flows.trigger(
    flow_id="your-flow-id",
    input_data={
        "key": "value",
        "data": "your data here"
    }
)

print(response)

Trigger Workflow with File

Upload a file

with open("/path/to/file.pdf", "rb") as f:
    response = client.flows.trigger_with_file(
        flow_id="your-flow-id",
        file=f,
        data={"note": "process this file"}
    )

Download from URL

response = client.flows.trigger_with_file(
    flow_id="your-flow-id",
    url="https://example.com/file.pdf",
    data={"note": "process downloaded file"}
)

Get Workflow Metrics

response = client.flows.get_metrics(
    start_date="2024-01-01",
    end_date="2024-02-01",
    status="completed"
)

print(f"Total workflows: {response['metrics']['total_workflows']}")
print(f"Completed: {response['metrics']['completed_workflows']}")
print(f"Failed: {response['metrics']['failed_workflows']}")
print(f"Average duration: {response['metrics']['avg_duration_ms']}ms")

Storage Operations

Upload a File

with open("/path/to/file.png", "rb") as f:
    response = client.storage.upload(file=f, path="uploads/images")

print("File ID:", response["fileId"])
print("URL:", response["url"])

Fetch a File

By File ID

response = client.storage.fetch("file-id-123")
print("Download URL:", response["url"])

By File Path

response = client.storage.fetch_by_path("uploads/images/1700000000000-file.png")
print("Download URL:", response["url"])

Delete a File

response = client.storage.delete("file-id-123")
print(response["message"])

Error Handling

The SDK provides specific exception classes for different error types.

from worqhat import Worqhat, WorqhatError, APIError, AuthenticationError

client = Worqhat()

try:
    response = client.db.execute_query(query="SELECT * FROM users")
except AuthenticationError as e:
    print(f"Authentication failed: {e.message}")
except APIError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.code}")
except WorqhatError as e:
    print(f"General error: {e.message}")

Exception Types

• WorqhatError - Base exception for all SDK errors
• APIError - Raised when the API returns an error response
• AuthenticationError - Raised when authentication fails
• ValidationError - Raised when request validation fails
• NotFoundError - Raised when a resource is not found
• RateLimitError - Raised when rate limit is exceeded

Type Hints

The SDK includes full type hints for better IDE support and type checking.

from worqhat import Worqhat
from typing import Dict, Any, List

client = Worqhat()

# Response is typed as Dict[str, Any]
response: Dict[str, Any] = client.db.execute_query(query="SELECT * FROM users")

# Access typed data
data: List[Dict[str, Any]] = response["data"]
execution_time: int = response["executionTime"]

Advanced Usage

Custom Timeout

client = Worqhat(api_key="your-api-key", timeout=120)  # 2 minutes

Backend Target

# Route to alternate backend
client = Worqhat(api_key="your-api-key", target="fyi")

Close Session Manually

client = Worqhat()
try:
    response = client.db.execute_query(query="SELECT * FROM users")
finally:
    client.close()

Examples

See the examples/ directory for complete working examples:

• basic_query.py - Basic database queries
• insert_update.py - Insert and update operations
• vector_search.py - Semantic and hybrid search
• workflows.py - Workflow operations
• storage.py - File storage operations

Requirements

• Python >= 3.8
• requests >= 2.31.0

Development

Install for Development

git clone https://github.com/worqhat/worqhat-python
cd worqhat-python
pip install -e ".[dev]"

Run Tests

pytest

Type Checking

mypy worqhat

Code Formatting

black worqhat

License

MIT

Support

For API support, contact [e-mail] or visit worqhat.com.