worqhat 4.2.0
Python SDK for Worqhat API
License: MIT
Requires Python: >=3.8
Keywords: worqhat, api, sdk, database, workflows

# Worqhat Python SDK

[![PyPI version](https://badge.fury.io/py/worqhat.svg)](https://pypi.org/project/worqhat/)
[![CircleCI](https://circleci.com/gh/worqhat/worqhat-api.svg?style=shield)](https://circleci.com/gh/worqhat/worqhat-api)
[![Known Vulnerabilities](https://snyk.io/test/github/worqhat/worqhat-api/badge.svg?targetFile=python-sdk/pyproject.toml)](https://snyk.io/test/github/worqhat/worqhat-api?targetFile=python-sdk/pyproject.toml)
[![Security: bandit](https://img.shields.io/badge/security-bandit-yellow.svg)](https://github.com/PyCQA/bandit)
[![Python Versions](https://img.shields.io/pypi/pyversions/worqhat.svg)](https://pypi.org/project/worqhat/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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

```bash
pip install worqhat
```

## Quick Start

```python
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

```python
from worqhat import Worqhat

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

### Advanced Configuration

```python
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

```python
import os

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

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

### Context Manager

```python
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

```python
response = client.db.execute_query(
    query="SELECT * FROM users WHERE email = {email} AND status = {status}",
    params={"email": "user@example.com", "status": "active"},
    environment="production"  # optional
)

print(response["data"])
```

#### Positional Parameters

```python
response = client.db.execute_query(
    query="SELECT * FROM users WHERE email = $1 AND status = $2",
    params=["user@example.com", "active"]
)

print(response["data"])
```

### Insert Data

Insert single or multiple records into a table.

#### Single Record

```python
response = client.db.insert(
    table="users",
    data={
        "email": "newuser@example.com",
        "name": "New User",
        "status": "active"
    }
)

print(response["data"])
```

#### Multiple Records

```python
response = client.db.insert(
    table="users",
    data=[
        {"email": "user1@example.com", "name": "User 1"},
        {"email": "user2@example.com", "name": "User 2"}
    ]
)

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

### Update Data

```python
response = client.db.update(
    table="users",
    data={"status": "inactive"},
    where={"email": "user@example.com"}
)

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

### Delete Data

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

print(response["message"])
```

### List Tables

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

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

### Get Table Schema

```python
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

```python
# 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.

```python
response = client.db.batch(
    operations=[
        {
            "type": "insert",
            "table": "users",
            "data": {"email": "user1@example.com", "name": "User 1"}
        },
        {
            "type": "update",
            "table": "users",
            "data": {"status": "active"},
            "where": {"email": "user1@example.com"}
        },
        {
            "type": "query",
            "query": "SELECT * FROM users WHERE email = $1",
            "params": ["user1@example.com"]
        }
    ],
    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

```python
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

```python
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.

```python
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.

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

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

#### By File Path

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

### Delete a File

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

## Error Handling

The SDK provides specific exception classes for different error types.

```python
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.

```python
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

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

### Backend Target

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

### Close Session Manually

```python
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

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

### Run Tests

```bash
pytest
```

### Type Checking

```bash
mypy worqhat
```

### Code Formatting

```bash
black worqhat
```

## License

MIT

## Support

For API support, contact support@worqhat.com or visit [worqhat.com](https://worqhat.com).