denser-retriever-sdk 0.1.0
Python SDK for Denser Retriever
Requires Python: >=3.8
Keywords: denser, retriever, search, knowledge-base, sdk

﻿# Denser Retriever SDK for Python

The official Python SDK for Denser Retriever Platform. Build powerful semantic search and retrieval applications with an intuitive API for knowledge base management, document ingestion, and intelligent querying.

## Table of Contents

- [Installation](#installation)
- [Quick Start](#quick-start)
- [Configuration](#configuration)
- [API Reference](#api-reference)
  - [Account Methods](#account-methods)
  - [Knowledge Base Methods](#knowledge-base-methods)
  - [Document Methods](#document-methods)
  - [Query Methods](#query-methods)
- [Error Handling](#error-handling)

## Installation

Install the SDK via pip:

```bash
pip install .
```

## Quick Start

```python
from denser_retriever import DenserRetriever

client = DenserRetriever(api_key="your-api-key")

def quick_example():
    # Create a knowledge base
    kb = client.create_knowledge_base("My First KB")
    kb_id = kb["data"]["id"]

    # Import text content
    client.import_text_content_and_poll(
        knowledge_base_id=kb_id,
        title="Getting Started",
        content="Denser Retriever enables semantic search across your documents."
    )

    # Search
    results = client.query(
        query="semantic search",
        knowledge_base_ids=[kb_id],
        limit=5
    )

    print(results["data"])

    # Cleanup
    client.delete_knowledge_base(kb_id)
```

## Configuration

Initialize the client with your API credentials:

```python
from denser_retriever import DenserRetriever

client = DenserRetriever(
    api_key="YOUR_API_KEY",           # Required: Your API key
    timeout=30                        # Optional: Request timeout in seconds (default: 30)
)
```

## API Reference

All methods return a dictionary with `success` (bool) and `data` fields.

### Account Methods

#### `get_usage() -> ApiResponse`

Retrieve current usage statistics for your organization.

```python
usage = client.get_usage()
print(f"Knowledge Bases: {usage['data']['knowledgeBaseCount']}")
print(f"Storage Used: {usage['data']['storageUsed']} bytes")
```

**Response:**

```python
{
    "success": True,
    "data": {
        "knowledgeBaseCount": int,
        "storageUsed": int  # bytes
    }
}
```

#### `get_balance() -> ApiResponse`

Retrieve current credit balance for your account.

```python
balance = client.get_balance()
print(f"Balance: {balance['data']['balance']} credits")
```

**Response:**

```python
{
    "success": True,
    "data": {
        "balance": float
    }
}
```

---

### Knowledge Base Methods

#### `create_knowledge_base(name: str, description: Optional[str] = None) -> ApiResponse`

Create a new knowledge base.

**Parameters:**

- `name` (str) - Knowledge base name (required)
- `description` (str, optional) - Optional description

```python
kb = client.create_knowledge_base(
    name="Technical Documentation",
    description="Product docs and API references"
)
kb_id = kb["data"]["id"]
```

**Response:**

```python
{
    "success": True,
    "data": {
        "id": str,
        "name": str,
        "description": str | None,
        "createdAt": str,
        "updatedAt": str
    }
}
```

#### `list_knowledge_bases() -> ApiResponse`

List all knowledge bases in your organization.

```python
kbs = client.list_knowledge_bases()
for kb in kbs['data']:
    print(f"{kb['name']} (ID: {kb['id']})")
```

#### `update_knowledge_base(knowledge_base_id: str, name: Optional[str] = None, description: Optional[str] = None) -> ApiResponse`

Update knowledge base metadata.

**Parameters:**

- `knowledge_base_id` (str) - Knowledge base ID (required)
- `name` (str, optional) - New name
- `description` (str, optional) - New description

```python
client.update_knowledge_base(
    knowledge_base_id=kb_id,
    name="Updated Name",
    description="Updated description"
)
```

#### `delete_knowledge_base(knowledge_base_id: str) -> ApiResponse`

Permanently delete a knowledge base and all its documents.

```python
client.delete_knowledge_base(kb_id)
```

---

### Document Methods

#### File Upload Workflow

Uploading files requires a three-step process:

##### Step 1: Get Presigned URL

#### `presign_upload_url(knowledge_base_id: str, file_name: str, size: int) -> ApiResponse`

Generate a presigned S3 URL for file upload.

**Parameters:**

- `knowledge_base_id` (str) - Target knowledge base ID
- `file_name` (str) - File name with extension
- `size` (int) - File size in bytes (max: 52,428,800)

```python
presign = client.presign_upload_url(kb_id, "document.pdf", 1024000)
file_id = presign["data"]["fileId"]
upload_url = presign["data"]["uploadUrl"]
expires_at = presign["data"]["expiresAt"]
```

##### Step 2: Upload File to S3

Use the requests library to PUT the file to the presigned URL:

```python
import requests

with open(file_path, "rb") as f:
    requests.put(upload_url, data=f, headers={"Content-Type": "application/octet-stream"})
```

##### Step 3: Register and Process

#### `import_file(file_id: str) -> ApiResponse`

Register the uploaded file for processing.

```python
doc = client.import_file(file_id)
print(f"Document ID: {doc['data']['id']}, Status: {doc['data']['status']}")
```

#### `import_file_and_poll(file_id: str, options: Optional[PollOptions] = None) -> ApiResponse`

Register file and automatically poll until processing completes.

**Parameters:**

- `file_id` (str) - File ID from presign_upload_url
- `options` (dict, optional) - Polling options
  - `intervalMs` (int) - Polling interval in milliseconds (default: 2000)
  - `timeoutMs` (int) - Maximum wait time in milliseconds (default: 600000)

```python
doc = client.import_file_and_poll(
    file_id,
    options={"intervalMs": 2000, "timeoutMs": 600000}
)
# Returns when status is "processed" or raises exception on "failed"/"timeout"
```

---

#### Text Content Import

#### `import_text_content(knowledge_base_id: str, title: str, content: str) -> ApiResponse`

Import text content directly as a document.

**Parameters:**

- `knowledge_base_id` (str) - Target knowledge base ID
- `title` (str) - Document title (max: 256 chars)
- `content` (str) - Text content (max: 1,000,000 chars)

```python
doc = client.import_text_content(
    knowledge_base_id=kb_id,
    title="API Documentation",
    content="Complete API reference and usage examples..."
)
```

#### `import_text_content_and_poll(knowledge_base_id: str, title: str, content: str, options: Optional[PollOptions] = None) -> ApiResponse`

Import text content and poll until processing completes.

```python
doc = client.import_text_content_and_poll(
    knowledge_base_id=kb_id,
    title="Release Notes",
    content="Version 2.0 includes...",
    options={"intervalMs": 1000, "timeoutMs": 300000}
)
```

---

#### Document Management

#### `list_documents(knowledge_base_id: str) -> ApiResponse`

List all documents in a knowledge base.

```python
docs = client.list_documents(kb_id)
for doc in docs['data']:
    print(f"{doc['title']} - {doc['status']} ({doc['size']} bytes)")
```

**Response:**

```python
{
    "success": True,
    "data": [
        {
            "id": str,
            "title": str,
            "type": str,
            "size": int,
            "status": str,  # "pending" | "processing" | "processed" | "failed" | "timeout"
            "createdAt": str
        }
    ]
}
```

#### `get_document_status(document_id: str) -> ApiResponse`

Check document processing status.

```python
status = client.get_document_status(doc_id)
print(f"Status: {status['data']['status']}")
```

**Status Values:**

- `pending` - Queued for processing
- `processing` - Currently being processed
- `processed` - Successfully processed and searchable
- `failed` - Processing failed
- `timeout` - Processing timed out

#### `delete_document(document_id: str) -> ApiResponse`

Permanently delete a document.

```python
client.delete_document(doc_id)
```

---

### Query Methods

#### `query(query: str, knowledge_base_ids: Optional[List[str]] = None, limit: Optional[int] = None) -> ApiResponse`

Perform semantic search across knowledge bases.

**Parameters:**

- `query` (str) - Search query (required, 1-8192 chars)
- `knowledge_base_ids` (list[str], optional) - Filter by specific knowledge bases
- `limit` (int, optional) - Maximum results (default: 10, max: 50)

```python
# Search across all knowledge bases
results = client.query("machine learning algorithms")

# Search within specific knowledge bases
results = client.query(
    query="deployment guide",
    knowledge_base_ids=[kb_id1, kb_id2],
    limit=20
)

# Process results
for item in results['data']:
    print(f"Score: {item['score']:.3f}")
    print(f"Title: {item['title']}")
    print(f"Content: {item['content']}")
    print(f"Document: {item['document_id']}")
    print(f"KB: {item['knowledge_base_id']}")
    print(f"Metadata: {item.get('metadata')}")
    print("---")
```

**Response:**

```python
{
    "success": True,
    "data": [
        {
            "id": str,
            "score": float,
            "document_id": str,
            "knowledge_base_id": str,
            "title": str,
            "type": str,
            "content": str,
            "metadata": {
                "source": str | None,
                "annotations": str
            }
        }
    ]
}
```

## Error Handling

The SDK raises `APIError` for all API-related errors, providing structured error information for robust error handling.

### APIError Class

```python
class APIError(Exception):
    def __init__(self, message: str, code: Optional[str] = None, 
                 http_status: Optional[int] = None, data: Optional[dict] = None):
        self.message = message      # Human-readable error message
        self.code = code            # Machine-readable error code
        self.http_status = http_status  # HTTP status code
        self.data = data            # Additional error context
```

### Error Handling Pattern

```python
from denser_retriever import DenserRetriever, APIError

client = DenserRetriever(api_key="your-api-key")

try:
    results = client.query(
        query="search term",
        knowledge_base_ids=["invalid-kb-id"],
        limit=5
    )
    print(results["data"])
except APIError as error:
    print(f"[{error.code}] {error.message}")
    print(f"HTTP Status: {error.http_status}")
    
    # Handle specific error codes
    if error.code == "INSUFFICIENT_CREDITS":
        print("Please top up your account to continue.")
    elif error.code == "NOT_FOUND":
        print("The requested resource does not exist.")
    elif error.code == "STORAGE_LIMIT_EXCEEDED":
        print(f"Storage quota exceeded: {error.data}")
    elif error.code == "KNOWLEDGE_BASE_LIMIT_EXCEEDED":
        print("Maximum knowledge bases reached.")
    else:
        print(f"An error occurred: {error.message}")
except Exception as error:
    print(f"Unexpected error: {error}")
```

### Common Error Codes

| Error Code | HTTP Status | Description |
| --------- | ----------- | ----------- |
| `INPUT_VALIDATION_FAILED` | 422 | Request parameters failed validation |
| `UNAUTHORIZED` | 401 | Invalid or missing API key |
| `FORBIDDEN` | 403 | Access to resource is denied |
| `NOT_FOUND` | 404 | Requested resource does not exist |
| `INSUFFICIENT_CREDITS` | 403 | Account has insufficient credits |
| `STORAGE_LIMIT_EXCEEDED` | 403 | Storage quota exceeded |
| `KNOWLEDGE_BASE_LIMIT_EXCEEDED` | 403 | Maximum knowledge bases reached |
| `INTERNAL_SERVER_ERROR` | 500 | Server encountered an error |