Denser Retriever

Python

Official Python SDK for Denser Retriever Platform

Installation

pip install denserai-retriever-sdk

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

    print(results["data"])

    # Cleanup
    client.delete_knowledge_base(kb_id)

if __name__ == "__main__":
    quick_example()

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)
)

You can find your API key in the Denser Retriever Dashboard.

API Reference

All methods return a dictionary with success (bool) and data fields.

Knowledge Base Management

create_knowledge_base

Create a new knowledge base.

Parameters:

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

Returns:

{
    "success": True,
    "data": {
        "id": str,
        "name": str,
        "description": str | None,
        "createdAt": str,
        "updatedAt": str
    }
}

list_knowledge_bases

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

Update knowledge base metadata.

Parameters:

ParameterTypeRequiredDescription
knowledge_base_idstrYesKnowledge base ID
namestrNoNew name
descriptionstrNoNew description
client.update_knowledge_base(
    knowledge_base_id=kb_id,
    name="Updated Name",
    description="Updated description"
)

delete_knowledge_base

Permanently delete a knowledge base and all its documents.

client.delete_knowledge_base(kb_id)

File Import Workflow

Uploading files requires a three-step process:

Step 1: Get Presigned URL

presign_upload_url

Generate a presigned URL for file upload.

Parameters:

ParameterTypeRequiredDescription
knowledge_base_idstrYesTarget knowledge base ID
file_namestrYesFile name with extension
sizeintYesFile 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

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"}
    )
import urllib.request

with open(file_path, "rb") as f:
    req = urllib.request.Request(
        upload_url,
        data=f,
        headers={"Content-Type": "application/octet-stream"},
        method="PUT"
    )
    with urllib.request.urlopen(req) as response:
        pass

The presigned URL expires after a certain time. Make sure to upload the file before expiration.

Step 3: Register and Process

import_file

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

Register file and automatically poll until processing completes.

Parameters:

ParameterTypeDefaultDescription
file_idstr-File ID from presign_upload_url
options.intervalMsint2000Polling interval in milliseconds
options.timeoutMsint600000Maximum wait time in milliseconds
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

Import text content directly as a document.

Parameters:

ParameterTypeRequiredDescription
knowledge_base_idstrYesTarget knowledge base ID
titlestrYesDocument title (max: 256 chars)
contentstrYesText 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

Import text content and poll until processing completes.

Parameters:

ParameterTypeDefaultDescription
knowledge_base_idstr-Target knowledge base ID
titlestr-Document title
contentstr-Text content
options.intervalMsint2000Polling interval in milliseconds
options.timeoutMsint600000Maximum wait time in milliseconds
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

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

Returns:

{
    "success": True,
    "data": [
        {
            "id": str,
            "title": str,
            "type": str,
            "size": int,
            "status": str,  # "pending" | "processing" | "processed" | "failed" | "timeout"
            "createdAt": str
        }
    ]
}

get_document_status

Check document processing status.

status = client.get_document_status(doc_id)
print(f"Status: {status['data']['status']}")

Status Values:

StatusDescription
pendingQueued for processing
processingCurrently being processed
processedSuccessfully processed and searchable
failedProcessing failed
timeoutProcessing timed out

delete_document

Permanently delete a document.

client.delete_document(doc_id)

Query Method

query

Perform semantic search across knowledge bases.

Parameters:

ParameterTypeRequiredDescription
querystrYesSearch query (1-8192 chars)
knowledge_base_idslist[str]NoFilter by specific knowledge bases
limitintNoMaximum 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("---")

Returns:

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

Account Management

get_usage

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

Returns:

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

get_balance

Retrieve current credit balance for your account.

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

Returns:

{
    "success": True,
    "data": {
        "balance": float
    }
}

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: str | None = None, 
                 http_status: int | None = None, data: dict | None = 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 CodeHTTP StatusDescription
INPUT_VALIDATION_FAILED422Request parameters failed validation
UNAUTHORIZED401Invalid or missing API key
FORBIDDEN403Access to resource is denied
NOT_FOUND404Requested resource does not exist
INSUFFICIENT_CREDITS403Account has insufficient credits
STORAGE_LIMIT_EXCEEDED403Storage quota exceeded
KNOWLEDGE_BASE_LIMIT_EXCEEDED403Maximum knowledge bases reached
INTERNAL_SERVER_ERROR500Server encountered an error

Always implement proper error handling in your application to gracefully handle these error cases.

Next Steps

On this page