Development Guide
This guide provides best practices and patterns for integrating the Vedaya API into your applications, covering authentication, error handling, performance optimization, and production deployment.
API Authentication
API Keys
All requests to the Vedaya API require authentication using an API key:
headers = {
'Authorization': 'Bearer YOUR_API_KEY'
}
Best Practices for API Key Management
- Never expose your API key in client-side code - Keep it secure in server-side code or environment variables
- Use different API keys for development and production - This helps isolate issues and manage usage
- Implement key rotation - Periodically generate new API keys and deprecate old ones
- Set up proper access controls - Limit which team members have access to production keys
Example using environment variables in Node.js:
// Load from .env file
require('dotenv').config();
const headers = {
'Authorization': `Bearer ${process.env.VEDAYA_API_KEY}`
};
Error Handling
The Vedaya API uses standard HTTP status codes and returns detailed error messages to help troubleshoot issues.
Common Status Codes
200
: Success
400
: Bad Request - Check your request parameters
401
: Unauthorized - Invalid or missing API key
404
: Not Found - Resource doesn’t exist
429
: Too Many Requests - Rate limit exceeded
500
: Internal Server Error - Something went wrong on our side
Implementing Robust Error Handling
import requests
def make_api_call(endpoint, method='GET', data=None, files=None):
url = f"https://vedaya-backend.fly.dev{endpoint}"
headers = {
'Authorization': f'Bearer {API_KEY}'
}
try:
if method == 'GET':
response = requests.get(url, headers=headers, params=data)
elif method == 'POST':
response = requests.post(url, headers=headers, json=data, files=files)
# Check if the request was successful
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as http_err:
# Handle HTTP errors
if response.status_code == 401:
print("Authentication error. Check your API key.")
elif response.status_code == 429:
print("Rate limit exceeded. Please retry after some time.")
else:
print(f"HTTP error occurred: {http_err}")
print(f"Response: {response.text}")
return None
except requests.exceptions.ConnectionError:
print("Connection error. Please check your internet connection.")
return None
except requests.exceptions.Timeout:
print("Request timed out. Please try again.")
return None
except requests.exceptions.RequestException as err:
print(f"An error occurred: {err}")
return None
Rate Limiting and Throttling
Vedaya implements rate limiting to ensure fair usage of the API. When you exceed the rate limit, you’ll receive a 429 Too Many Requests
response.
Best Practices for Handling Rate Limits
- Implement exponential backoff - Gradually increase wait time between retries
- Cache responses when possible - Reduce the number of API calls
- Batch requests - Process multiple documents in a single request
- Monitor your usage - Keep track of your API usage to avoid hitting limits
import time
import random
def call_api_with_backoff(endpoint, max_retries=5):
retries = 0
while retries < max_retries:
response = requests.get(f"https://vedaya-backend.fly.dev{endpoint}", headers=headers)
if response.status_code == 200:
return response.json()
if response.status_code == 429:
# Apply exponential backoff with jitter
sleep_time = (2 ** retries) + random.random()
print(f"Rate limit exceeded. Retrying in {sleep_time:.2f} seconds...")
time.sleep(sleep_time)
retries += 1
else:
# Handle other errors
print(f"Error: {response.status_code}")
print(response.text)
break
return None
Asynchronous Operations
Many Vedaya API operations (document processing, entity detection) are asynchronous. They initiate a process and return immediately, with the actual processing happening in the background.
Pattern for Handling Asynchronous Operations
import time
def process_document(file_path):
# Step 1: Upload the document
files = [('files', (file_path, open(file_path, 'rb')))]
response = requests.post("https://vedaya-backend.fly.dev/api/data-ingestion/upload-files",
headers=headers,
files=files)
file_id = response.json()['file_ids'][0]
# Step 2: Poll for processing status
while True:
status_response = requests.get(f"https://vedaya-backend.fly.dev/api/data-ingestion/status/{file_id}",
headers=headers)
status = status_response.json()['status']
if status == 'completed':
print("Document processing completed!")
break
elif status == 'failed':
print("Document processing failed.")
print(status_response.json()['error'])
break
else:
print(f"Processing status: {status}. Checking again in 5 seconds...")
time.sleep(5)
return file_id
Optimizing Document Processing
- Optimize file sizes - Compress images, remove unnecessary content
- Chunk appropriately - Adjust chunk size based on your content type
- Batch process documents - Process multiple documents in parallel
- Use appropriate PDF extractors - Choose the right tool for your document type
Optimizing Queries
- Be specific with queries - More specific queries yield better results
- Adjust top_k parameter - Only retrieve as many chunks as needed
- Cache common queries - Store results for frequently asked questions
- Use vector filtering - When available, filter by metadata to improve relevance
Webhook Integration
For production applications, consider using webhooks to receive notifications when asynchronous processes complete:
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/webhook/document-processed', methods=['POST'])
def document_processed_webhook():
data = request.json
# Verify webhook signature (implement your own security check)
if not verify_webhook_signature(request):
return jsonify({"status": "error", "message": "Invalid signature"}), 401
# Process the webhook data
file_id = data['file_id']
status = data['status']
if status == 'completed':
# Document processing completed successfully
# Trigger your business logic here
pass
elif status == 'failed':
# Document processing failed
# Implement error handling
error_message = data.get('error', 'Unknown error')
print(f"Processing failed for file {file_id}: {error_message}")
return jsonify({"status": "success"})
if __name__ == '__main__':
app.run(debug=True, port=5000)
Testing and Debugging
Testing Strategies
- Create a test suite - Automated tests for different API operations
- Use mock data - Test with consistent sample documents
- Test error scenarios - Ensure your app handles API errors gracefully
- Perform integration testing - Test the complete workflow
- API request logging - Log all API requests and responses
- Response inspection - Examine response bodies for error details
- Status polling - Check process status for asynchronous operations
Production Deployment Checklist
Before deploying to production:
- Security audit - Ensure API keys are securely stored
- Error handling - Implement comprehensive error handling
- Rate limit management - Add backoff and retry logic
- Monitoring - Set up alerts for API failures
- Backups - Plan for data backup and recovery
- Scalability - Design for increased load
- Testing - Perform thorough testing of all API integrations
Integration Best Practices by Framework
Node.js/Express
const axios = require('axios');
require('dotenv').config();
const vedayaAPI = axios.create({
baseURL: 'https://vedaya-backend.fly.dev',
headers: {
'Authorization': `Bearer ${process.env.VEDAYA_API_KEY}`,
'Content-Type': 'application/json'
}
});
// Add request interceptor for logging
vedayaAPI.interceptors.request.use(request => {
console.log('Starting Request', request.method, request.url);
return request;
});
// Add response interceptor for error handling
vedayaAPI.interceptors.response.use(
response => response,
error => {
console.error('API Error:', error.response ? error.response.data : error.message);
return Promise.reject(error);
}
);
module.exports = vedayaAPI;
Python/FastAPI
import os
from fastapi import FastAPI, HTTPException, Depends
import httpx
from functools import lru_cache
app = FastAPI()
class VedayaAPI:
def __init__(self, api_key: str):
self.base_url = "https://vedaya-backend.fly.dev"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.client = httpx.AsyncClient(base_url=self.base_url, headers=self.headers)
async def query_knowledge_base(self, query: str, top_k: int = 3):
response = await self.client.post("/api/retrieval/query", json={
"query": query,
"top_k": top_k
})
if response.status_code != 200:
raise HTTPException(status_code=response.status_code,
detail=f"Vedaya API error: {response.text}")
return response.json()
@lru_cache()
def get_vedaya_api():
api_key = os.getenv("VEDAYA_API_KEY")
if not api_key:
raise Exception("VEDAYA_API_KEY environment variable not set")
return VedayaAPI(api_key)
@app.get("/search")
async def search(query: str, vedaya_api: VedayaAPI = Depends(get_vedaya_api)):
return await vedaya_api.query_knowledge_base(query)