Python SDK
The official Python SDK for Hypersave with async support.
Installation
pip install hypersaveQuick Start
from hypersave import Hypersave
# Initialize client
client = Hypersave(api_key="hs_live_xxxxx")
# Save content
result = client.save(
content="Meeting notes: Discussed Q1 roadmap...",
title="Q1 Planning Meeting",
tags=["meetings", "planning"]
)
print(f"Saved document: {result.document_id}")
# Ask a question
answer = client.ask("What was discussed in the Q1 planning meeting?")
print(answer.text)Async Usage
import asyncio
from hypersave import AsyncHypersave
async def main():
client = AsyncHypersave(api_key="hs_live_xxxxx")
# Concurrent operations
results = await asyncio.gather(
client.save(content="Document 1"),
client.save(content="Document 2"),
client.ask("What documents do I have?")
)
print(results)
asyncio.run(main())Configuration
from hypersave import Hypersave
client = Hypersave(
api_key="hs_live_xxxxx",
# Optional configuration
base_url="https://api.hypersave.io", # Custom API endpoint
timeout=30.0, # Request timeout (seconds)
max_retries=3, # Auto-retry count
debug=False, # Enable debug logging
)Core Methods
save()
Save content to Hypersave.
from hypersave import Hypersave
client = Hypersave(api_key="...")
# Save text content
result = client.save(
content="Your content here",
title="Document Title",
tags=["tag1", "tag2"],
metadata={"source": "manual", "author": "john"}
)
# Save from URL
result = client.save(
url="https://example.com/article",
tags=["research"]
)
print(result.document_id) # doc_xxxask()
Ask questions about your memories.
# Basic question
result = client.ask("What is our Q1 budget?")
print(result.answer)
# With context
result = client.ask(
query="What is our Q1 budget?",
include_context=True,
limit=5
)
print(result.answer)
for ctx in result.context:
print(f"- {ctx.title}: {ctx.content[:100]}...")search()
Semantic search across memories.
# Basic search
results = client.search("machine learning")
for doc in results:
print(f"{doc.title} (score: {doc.score:.2f})")
print(f" {doc.content[:100]}...")
# Advanced search
results = client.search(
query="product roadmap",
limit=10,
threshold=0.7,
tags=["planning"],
hybrid=True # Semantic + keyword
)memories.list()
List all memories.
# List with pagination
memories = client.memories.list(limit=20, offset=0)
for memory in memories.data:
print(f"{memory.id}: {memory.title}")
print(f"Total: {memories.pagination.total}")memories.delete()
Delete a memory.
client.memories.delete("doc_xxx")Advanced Usage
Bulk Operations
# Bulk ingest
job = client.ingest(
items=[
{"content": "Document 1...", "tags": ["batch"]},
{"content": "Document 2...", "tags": ["batch"]},
{"url": "https://example.com/article"},
],
async_mode=True
)
# Check status
status = client.ingest.status(job.job_id)
print(f"Progress: {status.progress.percentage}%")Error Handling
from hypersave import Hypersave, HypersaveError, RateLimitError
client = Hypersave(api_key="...")
try:
result = client.save(content="...")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after}s")
except HypersaveError as e:
print(f"API error: {e.message}")Context Manager
from hypersave import Hypersave
with Hypersave(api_key="...") as client:
result = client.save(content="Hello!")
# Connection automatically closedPandas Integration
import pandas as pd
from hypersave import Hypersave
client = Hypersave(api_key="...")
# Export memories to DataFrame
memories = client.memories.list(limit=1000)
df = pd.DataFrame([m.dict() for m in memories.data])
# Analyze your memories
print(df.groupby('tags').count())Examples
FastAPI Integration
from fastapi import FastAPI, Depends
from hypersave import AsyncHypersave
app = FastAPI()
async def get_hypersave():
return AsyncHypersave(api_key="hs_live_xxxxx")
@app.post("/api/search")
async def search(
query: str,
hypersave: AsyncHypersave = Depends(get_hypersave)
):
results = await hypersave.search(query)
return {"results": results}
@app.post("/api/save")
async def save(
content: str,
hypersave: AsyncHypersave = Depends(get_hypersave)
):
result = await hypersave.save(content=content)
return {"document_id": result.document_id}LangChain Integration
from langchain.retrievers import BaseRetriever
from langchain.schema import Document
from hypersave import Hypersave
class HypersaveRetriever(BaseRetriever):
def __init__(self, api_key: str, **kwargs):
self.client = Hypersave(api_key=api_key)
super().__init__(**kwargs)
def _get_relevant_documents(self, query: str) -> list[Document]:
results = self.client.search(query, limit=5)
return [
Document(
page_content=r.content,
metadata={"id": r.id, "title": r.title, "score": r.score}
)
for r in results
]
# Use with LangChain
retriever = HypersaveRetriever(api_key="hs_live_xxxxx")
docs = retriever.get_relevant_documents("machine learning")Full API documentation available at GitHub (opens in a new tab).