denser-retriever-sdk 0.1.0

Python SDK for Denser Retriever

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
• Quick Start
• Configuration
• API Reference
  • Account Methods
  • Knowledge Base Methods
  • Document Methods
  • Query Methods
• Error Handling

Installation

Install the SDK via pip:

pip install .

Quick Start

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:

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.

usage = client.get_usage()
print(f"Knowledge Bases: {usage['data']['knowledgeBaseCount']}")
print(f"Storage Used: {usage['data']['storageUsed']} bytes")

Response:

{
    "success": True,
    "data": {
        "knowledgeBaseCount": int,
        "storageUsed": int  # bytes
    }
}

get_balance() -> ApiResponse

Retrieve current credit balance for your account.

balance = client.get_balance()
print(f"Balance: {balance['data']['balance']} credits")

Response:

{
    "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

kb = client.create_knowledge_base(
    name="Technical Documentation",
    description="Product docs and API references"
)
kb_id = kb["data"]["id"]

Response:

{
    "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.

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

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.

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)

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:

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.

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)

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)

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.

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.

docs = client.list_documents(kb_id)
for doc in docs['data']:
    print(f"{doc['title']} - {doc['status']} ({doc['size']} bytes)")

Response:

{
    "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.

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.

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)

# 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:

{
    "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

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

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