Best Practices

Overview

This guide covers essential best practices for using the Perplexity SDKs in production environments. Following these practices will help you build robust, secure, and efficient applications.

Security Best Practices

Environment Variables

Always store API keys securely using environment variables:

Use environment variables

Store API keys in environment variables, never in source code.

import os
from perplexity import Perplexity

# Good: Use environment variables
client = Perplexity(
    api_key=os.environ.get("PERPLEXITY_API_KEY")
)

# Bad: Never hardcode API keys
# client = Perplexity(api_key="pplx-abc123...")  # DON'T DO THIS

import Perplexity from '@perplexity-ai/perplexity_ai';

// Good: Use environment variables
const client = new Perplexity({
    apiKey: process.env.PERPLEXITY_API_KEY
});

// Bad: Never hardcode API keys
// const client = new Perplexity({
//     apiKey: "pplx-abc123..."  // DON'T DO THIS
// });

Never commit API keys to version control. Use .env files locally and secure environment variable management in production.

Use .env files for local development

Create a .env file for local development (add it to .gitignore):

cat > .env << 'EOF'
PERPLEXITY_API_KEY=your_api_key_here
PERPLEXITY_MAX_RETRIES=3
PERPLEXITY_TIMEOUT=30000
EOF

from dotenv import load_dotenv
import os
from perplexity import Perplexity

# Load environment variables from .env file
load_dotenv()

client = Perplexity(
    api_key=os.getenv("PERPLEXITY_API_KEY"),
    max_retries=int(os.getenv("PERPLEXITY_MAX_RETRIES", "3"))
)

import dotenv from 'dotenv';
import Perplexity from '@perplexity-ai/perplexity_ai';

// Load environment variables from .env file
dotenv.config();

const client = new Perplexity({
    apiKey: process.env.PERPLEXITY_API_KEY,
    maxRetries: parseInt(process.env.PERPLEXITY_MAX_RETRIES || '3')
});

Validate environment variables

Check for required environment variables at startup.

import os
import sys
from perplexity import Perplexity

def create_client():
    api_key = os.getenv("PERPLEXITY_API_KEY")
    if not api_key:
        print("Error: PERPLEXITY_API_KEY environment variable is required")
        sys.exit(1)
    
    return Perplexity(api_key=api_key)

client = create_client()

import Perplexity from '@perplexity-ai/perplexity_ai';

function createClient(): Perplexity {
    const apiKey = process.env.PERPLEXITY_API_KEY;
    if (!apiKey) {
        console.error("Error: PERPLEXITY_API_KEY environment variable is required");
        process.exit(1);
    }
    
    return new Perplexity({ apiKey });
}

const client = createClient();

API Key Rotation

Implement secure API key rotation:

import os
import logging
from perplexity import Perplexity
from typing import Optional

class SecurePerplexityClient:
    def __init__(self, primary_key: Optional[str] = None, fallback_key: Optional[str] = None):
        self.primary_key = primary_key or os.getenv("PERPLEXITY_API_KEY")
        self.fallback_key = fallback_key or os.getenv("PERPLEXITY_API_KEY_FALLBACK")
        self.current_client = Perplexity(api_key=self.primary_key)
        self.logger = logging.getLogger(__name__)
    
    def _switch_to_fallback(self):
        """Switch to fallback API key if available"""
        if self.fallback_key:
            self.logger.warning("Switching to fallback API key")
            self.current_client = Perplexity(api_key=self.fallback_key)
            return True
        return False
    
    def search(self, query: str, **kwargs):
        try:
            return self.current_client.search.create(query=query, **kwargs)
        except Exception as e:
            if "authentication" in str(e).lower() and self._switch_to_fallback():
                return self.current_client.search.create(query=query, **kwargs)
            raise e

# Usage
client = SecurePerplexityClient()

import Perplexity from '@perplexity-ai/perplexity_ai';

class SecurePerplexityClient {
    private primaryKey: string;
    private fallbackKey?: string;
    private currentClient: Perplexity;
    
    constructor(primaryKey?: string, fallbackKey?: string) {
        this.primaryKey = primaryKey || process.env.PERPLEXITY_API_KEY!;
        this.fallbackKey = fallbackKey || process.env.PERPLEXITY_API_KEY_FALLBACK;
        this.currentClient = new Perplexity({ apiKey: this.primaryKey });
    }
    
    private switchToFallback(): boolean {
        if (this.fallbackKey) {
            console.warn("Switching to fallback API key");
            this.currentClient = new Perplexity({ apiKey: this.fallbackKey });
            return true;
        }
        return false;
    }
    
    async search(query: string, options?: any) {
        try {
            return await this.currentClient.search.create({ query, ...options });
        } catch (error: any) {
            if (error.message.toLowerCase().includes('authentication') && this.switchToFallback()) {
                return await this.currentClient.search.create({ query, ...options });
            }
            throw error;
        }
    }
}

// Usage
const client = new SecurePerplexityClient();

Rate Limiting and Efficiency

Intelligent Rate Limiting

Implement exponential backoff with jitter:

import time
import random
import asyncio
from typing import TypeVar, Callable, Any
import perplexity
from perplexity import Perplexity

T = TypeVar('T')

class RateLimitedClient:
    def __init__(self, client: Perplexity, max_retries: int = 5):
        self.client = client
        self.max_retries = max_retries
    
    def _calculate_delay(self, attempt: int) -> float:
        """Calculate delay with exponential backoff and jitter"""
        base_delay = 2 ** attempt
        jitter = random.uniform(0.1, 0.5)
        return min(base_delay + jitter, 60.0)  # Cap at 60 seconds
    
    def with_retry(self, func: Callable[[], T]) -> T:
        """Execute function with intelligent retry logic"""
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                return func()
            except perplexity.RateLimitError as e:
                last_exception = e
                if attempt < self.max_retries - 1:
                    delay = self._calculate_delay(attempt)
                    print(f"Rate limited. Retrying in {delay:.2f}s (attempt {attempt + 1})")
                    time.sleep(delay)
                    continue
                raise e
            except perplexity.APIConnectionError as e:
                last_exception = e
                if attempt < self.max_retries - 1:
                    delay = min(2 ** attempt, 10.0)  # Shorter delay for connection errors
                    print(f"Connection error. Retrying in {delay:.2f}s")
                    time.sleep(delay)
                    continue
                raise e
        
        raise last_exception
    
    def search(self, query: str, **kwargs):
        return self.with_retry(
            lambda: self.client.search.create(query=query, **kwargs)
        )

# Usage
client = RateLimitedClient(Perplexity())
result = client.search("artificial intelligence")

import Perplexity from '@perplexity-ai/perplexity_ai';

class RateLimitedClient {
    private client: Perplexity;
    private maxRetries: number;
    
    constructor(client: Perplexity, maxRetries: number = 5) {
        this.client = client;
        this.maxRetries = maxRetries;
    }
    
    private calculateDelay(attempt: number): number {
        const baseDelay = 2 ** attempt * 1000; // Convert to milliseconds
        const jitter = Math.random() * 500; // 0-500ms jitter
        return Math.min(baseDelay + jitter, 60000); // Cap at 60 seconds
    }
    
    async withRetry<T>(func: () => Promise<T>): Promise<T> {
        let lastError: any;
        
        for (let attempt = 0; attempt < this.maxRetries; attempt++) {
            try {
                return await func();
            } catch (error: any) {
                lastError = error;
                
                if (error.constructor.name === 'RateLimitError') {
                    if (attempt < this.maxRetries - 1) {
                        const delay = this.calculateDelay(attempt);
                        console.log(`Rate limited. Retrying in ${delay}ms (attempt ${attempt + 1})`);
                        await new Promise(resolve => setTimeout(resolve, delay));
                        continue;
                    }
                } else if (error.constructor.name === 'APIConnectionError') {
                    if (attempt < this.maxRetries - 1) {
                        const delay = Math.min(2 ** attempt * 1000, 10000);
                        console.log(`Connection error. Retrying in ${delay}ms`);
                        await new Promise(resolve => setTimeout(resolve, delay));
                        continue;
                    }
                }
                
                throw error;
            }
        }
        
        throw lastError;
    }
    
    async search(query: string, options?: any) {
        return this.withRetry(() => 
            this.client.search.create({ query, ...options })
        );
    }
}

// Usage
const client = new RateLimitedClient(new Perplexity());
const result = await client.search("artificial intelligence");

Request Batching

Efficiently batch multiple requests:

import asyncio
from typing import Callable, Awaitable, List, TypeVar, Generic
from perplexity import AsyncPerplexity, DefaultAioHttpClient

T = TypeVar('T')

class BatchProcessor(Generic[T]):
    def __init__(self, batch_size: int = 5, delay_between_batches: float = 1.0):
        self.batch_size = batch_size
        self.delay_between_batches = delay_between_batches
    
    async def process_batch(
        self, 
        items: List[str], 
        process_func: Callable[[str], Awaitable[T]]
    ) -> List[T]:
        """Process items in batches with rate limiting"""
        results = []
        
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            
            # Process batch concurrently
            tasks = [process_func(item) for item in batch]
            batch_results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Filter out exceptions and collect results
            for result in batch_results:
                if not isinstance(result, Exception):
                    results.append(result)
            
            # Delay between batches
            if i + self.batch_size < len(items):
                await asyncio.sleep(self.delay_between_batches)
        
        return results

# Usage
async def main():
    processor = BatchProcessor(batch_size=3, delay_between_batches=0.5)
    
    async with AsyncPerplexity(
        http_client=DefaultAioHttpClient()
    ) as client:
        
        async def search_query(query: str):
            return await client.search.create(query=query)
        
        queries = ["AI", "ML", "DL", "NLP", "CV"]
        results = await processor.process_batch(queries, search_query)
        
        print(f"Processed {len(results)} successful queries")

asyncio.run(main())

import Perplexity from '@perplexity-ai/perplexity_ai';

class BatchProcessor<T> {
    constructor(
        private batchSize: number = 5,
        private delayBetweenBatches: number = 1000
    ) {}
    
    async processBatch<R>(
        items: T[],
        processFunc: (item: T) => Promise<R>
    ): Promise<R[]> {
        const results: R[] = [];
        
        for (let i = 0; i < items.length; i += this.batchSize) {
            const batch = items.slice(i, i + this.batchSize);
            
            // Process batch concurrently
            const tasks = batch.map(item => 
                processFunc(item).catch(error => error)
            );
            const batchResults = await Promise.all(tasks);
            
            // Filter out exceptions and collect results
            for (const result of batchResults) {
                if (!(result instanceof Error)) {
                    results.push(result);
                }
            }
            
            // Delay between batches
            if (i + this.batchSize < items.length) {
                await new Promise(resolve => 
                    setTimeout(resolve, this.delayBetweenBatches)
                );
            }
        }
        
        return results;
    }
}

// Usage
async function main() {
    const processor = new BatchProcessor<string>(3, 500);
    const client = new Perplexity();
    
    const searchQuery = (query: string) => 
        client.search.create({ query });
    
    const queries = ["AI", "ML", "DL", "NLP", "CV"];
    const results = await processor.processBatch(queries, searchQuery);
    
    console.log(`Processed ${results.length} successful queries`);
}

main();

Production Configuration

Configuration Management

Use environment-based configuration for different deployment stages:

import os
from dataclasses import dataclass
from typing import Optional
import httpx
from perplexity import Perplexity, DefaultHttpxClient

@dataclass
class PerplexityConfig:
    api_key: str
    max_retries: int = 3
    timeout_seconds: float = 30.0
    max_connections: int = 100
    max_keepalive: int = 20
    environment: str = "production"
    
    @classmethod
    def from_env(cls) -> "PerplexityConfig":
        """Load configuration from environment variables"""
        api_key = os.getenv("PERPLEXITY_API_KEY")
        if not api_key:
            raise ValueError("PERPLEXITY_API_KEY environment variable is required")
        
        return cls(
            api_key=api_key,
            max_retries=int(os.getenv("PERPLEXITY_MAX_RETRIES", "3")),
            timeout_seconds=float(os.getenv("PERPLEXITY_TIMEOUT", "30.0")),
            max_connections=int(os.getenv("PERPLEXITY_MAX_CONNECTIONS", "100")),
            max_keepalive=int(os.getenv("PERPLEXITY_MAX_KEEPALIVE", "20")),
            environment=os.getenv("ENVIRONMENT", "production")
        )
    
    def create_client(self) -> Perplexity:
        """Create optimized client based on configuration"""
        timeout = httpx.Timeout(
            connect=5.0,
            read=self.timeout_seconds,
            write=10.0,
            pool=10.0
        )
        
        limits = httpx.Limits(
            max_keepalive_connections=self.max_keepalive,
            max_connections=self.max_connections,
            keepalive_expiry=60.0 if self.environment == "production" else 30.0
        )
        
        return Perplexity(
            api_key=self.api_key,
            max_retries=self.max_retries,
            timeout=timeout,
            http_client=DefaultHttpxClient(limits=limits)
        )

# Usage
config = PerplexityConfig.from_env()
client = config.create_client()

import Perplexity from '@perplexity-ai/perplexity_ai';
import https from 'https';

interface PerplexityConfig {
    apiKey: string;
    maxRetries: number;
    timeoutMs: number;
    maxConnections: number;
    maxKeepalive: number;
    environment: string;
}

class ConfigManager {
    static fromEnv(): PerplexityConfig {
        const apiKey = process.env.PERPLEXITY_API_KEY;
        if (!apiKey) {
            throw new Error("PERPLEXITY_API_KEY environment variable is required");
        }
        
        return {
            apiKey,
            maxRetries: parseInt(process.env.PERPLEXITY_MAX_RETRIES || '3'),
            timeoutMs: parseInt(process.env.PERPLEXITY_TIMEOUT || '30000'),
            maxConnections: parseInt(process.env.PERPLEXITY_MAX_CONNECTIONS || '100'),
            maxKeepalive: parseInt(process.env.PERPLEXITY_MAX_KEEPALIVE || '20'),
            environment: process.env.NODE_ENV || 'production'
        };
    }
    
    static createClient(config: PerplexityConfig): Perplexity {
        const httpsAgent = new https.Agent({
            keepAlive: true,
            keepAliveMsecs: config.environment === 'production' ? 60000 : 30000,
            maxSockets: config.maxConnections,
            maxFreeSockets: config.maxKeepalive,
            timeout: config.timeoutMs
        });
        
        return new Perplexity({
            apiKey: config.apiKey,
            maxRetries: config.maxRetries,
            timeout: config.timeoutMs,
            httpAgent: httpsAgent
        } as any);
    }
}

// Usage
const config = ConfigManager.fromEnv();
const client = ConfigManager.createClient(config);

Monitoring and Logging

Implement comprehensive monitoring:

import logging
import time
import functools
from typing import Any, Callable
from perplexity import Perplexity
import perplexity

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class MonitoredPerplexityClient:
    def __init__(self, client: Perplexity):
        self.client = client
        self.request_count = 0
        self.error_count = 0
        self.total_response_time = 0.0
    
    def _log_request(self, method: str, **kwargs):
        """Log request details"""
        self.request_count += 1
        logger.info(f"Making {method} request #{self.request_count}")
        logger.debug(f"Request parameters: {kwargs}")
    
    def _log_response(self, method: str, duration: float, success: bool = True):
        """Log response details"""
        self.total_response_time += duration
        avg_response_time = self.total_response_time / self.request_count
        
        if success:
            logger.info(f"{method} completed in {duration:.2f}s (avg: {avg_response_time:.2f}s)")
        else:
            self.error_count += 1
            logger.error(f"{method} failed after {duration:.2f}s (errors: {self.error_count})")
    
    def search(self, query: str, **kwargs):
        self._log_request("search", query=query, **kwargs)
        start_time = time.time()
        
        try:
            result = self.client.search.create(query=query, **kwargs)
            duration = time.time() - start_time
            self._log_response("search", duration, success=True)
            return result
        except Exception as e:
            duration = time.time() - start_time
            self._log_response("search", duration, success=False)
            logger.error(f"Search error: {type(e).__name__}: {e}")
            raise
    
    def get_stats(self):
        """Get client statistics"""
        return {
            "total_requests": self.request_count,
            "error_count": self.error_count,
            "success_rate": (self.request_count - self.error_count) / max(self.request_count, 1),
            "avg_response_time": self.total_response_time / max(self.request_count, 1)
        }

# Usage
client = MonitoredPerplexityClient(Perplexity())
result = client.search("machine learning")
print(client.get_stats())

import Perplexity from '@perplexity-ai/perplexity_ai';

interface ClientStats {
    totalRequests: number;
    errorCount: number;
    successRate: number;
    avgResponseTime: number;
}

class MonitoredPerplexityClient {
    private client: Perplexity;
    private requestCount: number = 0;
    private errorCount: number = 0;
    private totalResponseTime: number = 0;
    
    constructor(client: Perplexity) {
        this.client = client;
    }
    
    private logRequest(method: string, params: any): void {
        this.requestCount++;
        console.log(`Making ${method} request #${this.requestCount}`);
        console.debug(`Request parameters:`, params);
    }
    
    private logResponse(method: string, duration: number, success: boolean = true): void {
        this.totalResponseTime += duration;
        const avgResponseTime = this.totalResponseTime / this.requestCount;
        
        if (success) {
            console.log(`${method} completed in ${duration.toFixed(2)}ms (avg: ${avgResponseTime.toFixed(2)}ms)`);
        } else {
            this.errorCount++;
            console.error(`${method} failed after ${duration.toFixed(2)}ms (errors: ${this.errorCount})`);
        }
    }
    
    async search(query: string, options?: any) {
        this.logRequest("search", { query, ...options });
        const startTime = performance.now();
        
        try {
            const result = await this.client.search.create({ query, ...options });
            const duration = performance.now() - startTime;
            this.logResponse("search", duration, true);
            return result;
        } catch (error) {
            const duration = performance.now() - startTime;
            this.logResponse("search", duration, false);
            console.error(`Search error: ${error}`);
            throw error;
        }
    }
    
    getStats(): ClientStats {
        return {
            totalRequests: this.requestCount,
            errorCount: this.errorCount,
            successRate: (this.requestCount - this.errorCount) / Math.max(this.requestCount, 1),
            avgResponseTime: this.totalResponseTime / Math.max(this.requestCount, 1)
        };
    }
}

// Usage
const client = new MonitoredPerplexityClient(new Perplexity());
const result = await client.search("machine learning");
console.log(client.getStats());

Error Handling Best Practices

Graceful Degradation

Implement fallback strategies for different error types:

from typing import Optional, Dict, Any
import perplexity
from perplexity import Perplexity

class ResilientPerplexityClient:
    def __init__(self, client: Perplexity):
        self.client = client
        self.circuit_breaker_threshold = 5
        self.circuit_breaker_count = 0
        self.circuit_breaker_open = False
    
    def _should_circuit_break(self) -> bool:
        """Check if circuit breaker should be triggered"""
        return self.circuit_breaker_count >= self.circuit_breaker_threshold
    
    def _record_failure(self):
        """Record a failure for circuit breaker"""
        self.circuit_breaker_count += 1
        if self._should_circuit_break():
            self.circuit_breaker_open = True
            print("Circuit breaker activated - temporarily disabling API calls")
    
    def _record_success(self):
        """Record a success - reset circuit breaker"""
        self.circuit_breaker_count = 0
        self.circuit_breaker_open = False
    
    def search_with_fallback(
        self, 
        query: str, 
        fallback_response: Optional[Dict[str, Any]] = None
    ):
        """Search with graceful degradation"""
        if self.circuit_breaker_open:
            print("Circuit breaker open - returning fallback response")
            return fallback_response or {
                "query": query,
                "results": [],
                "status": "service_unavailable"
            }
        
        try:
            result = self.client.search.create(query=query)
            self._record_success()
            return result
            
        except perplexity.RateLimitError:
            print("Rate limited - implementing backoff strategy")
            # Could implement intelligent backoff here
            raise
            
        except perplexity.APIConnectionError as e:
            print(f"Connection error: {e}")
            self._record_failure()
            return fallback_response or {
                "query": query,
                "results": [],
                "status": "connection_error"
            }
            
        except Exception as e:
            print(f"Unexpected error: {e}")
            self._record_failure()
            return fallback_response or {
                "query": query,
                "results": [],
                "status": "error"
            }

# Usage
client = ResilientPerplexityClient(Perplexity())
result = client.search_with_fallback("machine learning")

import Perplexity from '@perplexity-ai/perplexity_ai';

interface FallbackResponse {
    query: string;
    results: any[];
    status: string;
}

class ResilientPerplexityClient {
    private client: Perplexity;
    private circuitBreakerThreshold: number = 5;
    private circuitBreakerCount: number = 0;
    private circuitBreakerOpen: boolean = false;
    
    constructor(client: Perplexity) {
        this.client = client;
    }
    
    private shouldCircuitBreak(): boolean {
        return this.circuitBreakerCount >= this.circuitBreakerThreshold;
    }
    
    private recordFailure(): void {
        this.circuitBreakerCount++;
        if (this.shouldCircuitBreak()) {
            this.circuitBreakerOpen = true;
            console.log("Circuit breaker activated - temporarily disabling API calls");
        }
    }
    
    private recordSuccess(): void {
        this.circuitBreakerCount = 0;
        this.circuitBreakerOpen = false;
    }
    
    async searchWithFallback(
        query: string,
        fallbackResponse?: FallbackResponse
    ): Promise<any> {
        if (this.circuitBreakerOpen) {
            console.log("Circuit breaker open - returning fallback response");
            return fallbackResponse || {
                query,
                results: [],
                status: "service_unavailable"
            };
        }
        
        try {
            const result = await this.client.search.create({ query });
            this.recordSuccess();
            return result;
            
        } catch (error: any) {
            if (error.constructor.name === 'RateLimitError') {
                console.log("Rate limited - implementing backoff strategy");
                throw error;
            } else if (error.constructor.name === 'APIConnectionError') {
                console.log(`Connection error: ${error.message}`);
                this.recordFailure();
                return fallbackResponse || {
                    query,
                    results: [],
                    status: "connection_error"
                };
            } else {
                console.log(`Unexpected error: ${error.message}`);
                this.recordFailure();
                return fallbackResponse || {
                    query,
                    results: [],
                    status: "error"
                };
            }
        }
    }
}

// Usage
const client = new ResilientPerplexityClient(new Perplexity());
const result = await client.searchWithFallback("machine learning");

Testing Best Practices

Unit Testing with Mocking

Create testable code with proper mocking:

import unittest
from unittest.mock import Mock, patch
from perplexity import Perplexity, RateLimitError
from perplexity.types.search_create_response import SearchCreateResponse, Result

class TestPerplexityIntegration(unittest.TestCase):
    def setUp(self):
        self.mock_client = Mock(spec=Perplexity)

    def test_search_success(self):
        # Mock successful response
        mock_result = Result(
            title="Test Result",
            url="https://example.com",
            snippet="Test snippet"
        )

        mock_response = SearchCreateResponse(
            id="search_123",
            results=[mock_result]
        )

        self.mock_client.search.create.return_value = mock_response

        # Test your application logic
        result = self.mock_client.search.create(query="Perplexity Search API rate limit and retry handling example")

        self.assertEqual(result.id, "search_123")
        self.assertEqual(len(result.results), 1)
        self.assertEqual(result.results[0].title, "Test Result")

    @patch('perplexity.Perplexity')
    def test_rate_limit_handling(self, mock_perplexity_class):
        # Mock rate limit error
        mock_client = Mock()
        mock_perplexity_class.return_value = mock_client
        mock_client.search.create.side_effect = RateLimitError(
            "Rate limited",
            response=Mock(status_code=429),
            body={}
        )

        # Test your error handling logic here
        with self.assertRaises(RateLimitError):
            mock_client.search.create(query="Perplexity Search API Python SDK quickstart example")

if __name__ == '__main__':
    unittest.main()

import { jest, describe, beforeEach, test, expect } from '@jest/globals';
import Perplexity from '@perplexity-ai/perplexity_ai';

// Mock the Perplexity client
jest.mock('@perplexity-ai/perplexity_ai');

describe('Perplexity Integration', () => {
    let mockClient: any;

    beforeEach(() => {
        mockClient = new Perplexity();
    });

    test('should handle successful search', async () => {
        const mockResponse = {
            id: "search_123",
            results: [{ title: "Test Result", url: "https://example.com", snippet: "Test snippet" }]
        };

        jest.spyOn(mockClient.search, 'create').mockResolvedValue(mockResponse as any);

        const result = await mockClient.search.create({ query: "Perplexity Search API rate limit and retry handling example" });

        expect(result.id).toBe("search_123");
        expect(result.results).toHaveLength(1);
        expect(result.results[0].title).toBe("Test Result");
    });

    test('should handle rate limit errors', async () => {
        const rateLimitError = new Error('Rate limited');

        jest.spyOn(mockClient.search, 'create').mockRejectedValue(rateLimitError);

        await expect(mockClient.search.create({ query: "Perplexity Search API Python SDK quickstart example" })).rejects.toThrow('Rate limited');
    });
});

Response — Perplexity Search API rate limit and retry handling example

{
  "id": "fc4a9702-2f8c-454c-bff5-04f63041aaa7",
  "results": [
    {
      "snippet": "",
      "title": "Best Practices",
      "url": "https://docs.perplexity.ai/docs/search/best-practices",
      "date": null,
      "last_updated": "2026-05-26"
    },
    {
      "snippet": "",
      "title": "Perplexity API: Setup, Models, Integration & Best Practices",
      "url": "https://zuplo.com/learning-center/perplexity-api",
      "date": "2025-03-28",
      "last_updated": "2026-05-24"
    },
    {
      "snippet": "I am getting a \" [429] Request rate limit exceeded, please try again later.\" error from make.com.\n...\nYou can try placing a sleep module before each ai module to wait 3-5 seconds.",
      "title": "How can i optimise the number of api calls, and cell ...",
      "url": "https://community.make.com/t/how-can-i-optimise-the-number-of-api-calls-and-cell-updates-in-this-scenario/49923",
      "date": "2024-08-07",
      "last_updated": "2026-05-16"
    },
    {
      "snippet": "## Search API Rate Limits\nThe Search API has separate rate limits that apply to all usage tiers:\n| Endpoint       | Rate Limit             | Burst Capacity |\n| -------------- | ---------------------- | -------------- |\n| POST `/search` | 50 requests per second | 50 requests    |\n**Search Rate Limiter Behavior:**\n* **Burst**: Can handle 50 requests instantly\n* **Sustained**: Exactly 50 QPS average over time\n<Note>\nSearch rate limits are independent of your usage tier and apply consistently across all accounts using the same leaky bucket algorithm.\n</Note>\n...\nWhen you exceed your rate limits:\n1. **429 Error** - Your request gets rejected with \"Too Many Requests\"\n2. **Continuous Refill** - Tokens refill continuously based on your rate limit\n3. **Immediate Recovery** - New requests become available as soon as tokens refill\n**Example Recovery Times:**\n* **50 QPS limit**: 1 token refills every 20ms\n* **500 QPS limit**: 1 token refills every 2ms\n* **1,000 QPS limit**: 1 token refills every 1ms\n...\nVisit the [API Platform console](https://console.perplexity.ai) to see your current tier and total spending.\n...\nAdd credits to your account through the billing section.\nYour tier will automatically upgrade once you reach the spending threshold.\n...\nYour new rate limits take effect immediately after the tier upgrade.\nCheck your settings page to confirm.\n...\nIf you require custom rate limits beyond Tier 5, [fill out our rate limit increase request form](https://perplexity.typeform.com/to/yctmfyVT) and we'll review your use case to accommodate your needs.",
      "title": "Rate Limits & Usage Tiers - Perplexity API",
      "url": "https://docs.perplexity.ai/docs/admin/rate-limits-usage-tiers",
      "date": null,
      "last_updated": "2026-05-26"
    },
    {
      "snippet": "",
      "title": "Best Practices",
      "url": "https://docs.perplexity.ai/docs/sdk/best-practices",
      "date": null,
      "last_updated": "2026-05-27"
    },
    {
      "snippet": "Use highly specific queries for more targeted results.\nFor example, instead of searching for “AI”, use a detailed query like “artificial intelligence machine learning healthcare applications 2024”.\n```\n# Better: Specific query\nsearch = client.search.create(\nquery=\"artificial intelligence medical diagnosis accuracy 2024\",\nmax_results=10\n)\n# Avoid: Vague query\nsearch = client.search.create(\nquery=\"AI medical\",\n...\n)\n```\n...\nBreak your main topic into related sub-queries to cover all aspects of your research.\nUse the multi-query search feature to run multiple related queries in a single request for more comprehensive and relevant information.\n```\n...\nclient = Perplexity()\n# Comprehensive research with related queries\nsearch = client.search.create(\n...\n)\n...\n```\nYou can include up to 5 queries in a single multi-query request for efficient batch processing.\n3\nHandle rate limits efficiently\nImplement exponential backoff for rate limit errors and use appropriate batching strategies.\n```\nimport time\nimport random\nfrom perplexity import RateLimitError\ndef search_with_retry(client, query, max_retries=3):\nfor attempt in range(max_retries):\ntry:\nreturn client.search.create(query=query)\nexcept RateLimitError:\nif attempt < max_retries - 1:\n# Exponential backoff with jitter\ndelay = (2 ** attempt) + random.uniform(0, 1)\ntime.sleep(delay)\nelse:\nraise\n# Usage\ntry:\nsearch = search_with_retry(client, \"AI developments\")\nfor result in search.results:\nprint(f\"{result.title}: {result.url}\")\nexcept RateLimitError:\nprint(\"Maximum retries exceeded for search\")\n```\n4\nProcess concurrent searches efficiently\nUse async for concurrent requests while respecting rate limits.\n```\nimport asyncio\nfrom perplexity import AsyncPerplexity\nasync def batch_search(queries, batch_size=3, delay_ms=1000):\nasync with AsyncPerplexity() as client:\nresults = []\nfor i in range(0, len(queries), batch_size):\nbatch = queries[i:i + batch_size]\nbatch_tasks = [\nclient.search.create(query=query, max_results=5)\nfor query in batch\n]\nbatch_results = await asyncio.gather(*batch_tasks)\nresults.extend(batch_results)\n# Add delay between batches\nif i + batch_size < len(queries):\nawait asyncio.sleep(delay_ms / 1000)\nreturn results\n# Usage\nqueries = [\"AI developments\", \"climate change\", \"space exploration\"]\nresults = asyncio.run(batch_search(queries))\nprint(f\"Processed {len(results)} searches\")\n```\n...\n#### Rate-Limited Concurrent Processing\nFor large-scale applications, implement controlled concurrency with rate limiting:\n```\nimport asyncio\nfrom perplexity import AsyncPerplexity\nclass SearchManager:\ndef __init__(self, max_concurrent=5, delay_between_batches=1.0):\nself.max_concurrent = max_concurrent\nself.delay_between_batches = delay_between_batches\nself.semaphore = asyncio.Semaphore(max_concurrent)\nasync def search_single(self, client, query):\nasync with self.semaphore:\nreturn await client.search.create(query=query, max_results=5)\nasync def search_many(self, queries):\nasync with AsyncPerplexity() as client:\ntasks = [\nself.search_single(client, query)\nfor query in queries\n]\nresults = await asyncio.gather(*tasks, return_exceptions=True)\n# Filter out exceptions and return successful results\nsuccessful_results = [\nresult for result in results\nif not isinstance(result, Exception)\n]\nreturn successful_results\n# Usage\nasync def main():\nmanager = SearchManager(max_concurrent=3)\nqueries = [\n\"AI research 2024\",\n\"quantum computing advances\",\n\"renewable energy innovations\",\n\"biotechnology breakthroughs\",\n\"space exploration updates\"\n]\nresults = await manager.search_many(queries)\nprint(f\"Successfully processed {len(results)} out of {len(queries)} searches\")\nasyncio.run(main())\n```\n#### Error Handling in Async Operations\nImplement robust error handling for async search operations:\n```\nimport asyncio\nimport logging\nfrom perplexity import AsyncPerplexity, APIStatusError, RateLimitError\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\nasync def resilient_search(client, query, max_retries=3):\nfor attempt in range(max_retries):\ntry:\nresult = await client.search.create(query=query, max_results=5)\nlogger.info(f\"Search successful for: {query}\")\nreturn result\nexcept RateLimitError as e:\nif attempt < max_retries - 1:\ndelay = 2 ** attempt\nlogger.warning(f\"Rate limited for '{query}', retrying in {delay}s\")\nawait asyncio.sleep(delay)\nelse:\nlogger.error(f\"Max retries exceeded for: {query}\")\nreturn None\nexcept APIStatusError as e:\nlogger.error(f\"API error for '{query}': {e}\")\nreturn None\nexcept Exception as e:\nlogger.error(f\"Unexpected error for '{query}': {e}\")\nreturn None\nasync def main():\nasync with AsyncPerplexity() as client:\nqueries = [\"AI developments\", \"invalid query\", \"tech trends\"]\ntasks = [resilient_search(client, query) for query in queries]\nresults = await asyncio.gather(*tasks)\nsuccessful_results = [r for r in results if r is not None]\nprint(f\"Successful searches: {len(successful_results)}/{len(queries)}\")\nasyncio.run(main())\n```\n...\nRequest only the number of results you actually need.\nMore results = longer response times.\n...\nImplement caching for queries that don’t need real-time results.",
      "title": "Best Practices - Perplexity",
      "url": "https://docs.perplexity.ai/guides/search-best-practices",
      "date": "2025-10-09",
      "last_updated": "2026-05-10"
    },
    {
      "snippet": "",
      "title": "Fix Perplexity API Errors 429, 405, 500 | Complete Tutorial 2025",
      "url": "https://www.hostingseekers.com/blog/fix-perplexity-api-errors-tutorial/",
      "date": "2025-11-21",
      "last_updated": "2026-05-16"
    }
  ],
  "server_time": null
}

Response — Perplexity Search API Python SDK quickstart example

{
  "id": "a3284d7c-f02c-45fc-82db-d368ad880edc",
  "results": [
    {
      "snippet": "",
      "title": "Quickstart - Perplexity API",
      "url": "https://docs.perplexity.ai/docs/getting-started/quickstart",
      "date": null,
      "last_updated": "2026-05-26"
    },
    {
      "snippet": "",
      "title": "Perplexity Search API",
      "url": "https://docs.perplexity.ai/docs/search/quickstart",
      "date": null,
      "last_updated": "2026-05-27"
    },
    {
      "snippet": "",
      "title": "Perplexity API",
      "url": "https://docs.perplexity.ai/docs/getting-started/overview",
      "date": null,
      "last_updated": "2026-05-26"
    },
    {
      "snippet": "",
      "title": "Perplexity Search APIdocs.perplexity.ai › guides › search-quickstart",
      "url": "https://docs.perplexity.ai/docs/search/quickstart.md",
      "date": null,
      "last_updated": "2026-05-27"
    },
    {
      "snippet": "### 2.\nSearch API Quickstart (Python & TypeScript SDKs)\nThe docs recommend using official SDKs for safety and type-safety; you can also call the HTTP endpoint directly (POST https://api.perplexity.ai/search) with an Authorization header.\nBelow is a minimal Python example that mirrors the documented pattern.\n#### Basic Python example (client.search.create)\n# Example (conceptual) — mirrors docs pattern\nfrom perplexity import Client  # hypothetical SDK import style\nclient = Client(api_key=”YOUR_API_KEY”)\nresp = client.search.create(\nquery=”latest AI model research 2025″,\nmax_results=5\n)\n# Example response shape (simplified):\n# resp.results -> [ { “title”: “…”, “url”: “…”, “snippet”: “…”, “rank”: 1 }, … ]\nprint(resp.results[0][“title”], resp.results[0][“url”])\nThis call returns ranked results you can present to users or feed into an LLM for grounded synthesis.\nIf you prefer raw HTTP the docs provide a curl example for POST /search.\n...\nStart with the quickstart, use multi-query for depth, control content with max_tokens_per_page, and organize keys and billing via API Groups.",
      "title": "Mastering the Perplexity AI API Documentation",
      "url": "https://seabuckdigital.com/perplexity-ai-api-documentation/",
      "date": "2025-11-08",
      "last_updated": "2026-04-08"
    },
  ],
  "server_time": null
}

Performance Best Practices Summary

Use environment variables for configuration

Never hardcode API keys or configuration values.

Implement intelligent rate limiting

Use exponential backoff with jitter for retry strategies.

Configure connection pooling

Optimize HTTP client settings for your use case.

Monitor and log appropriately

Track performance metrics and error rates.

Implement graceful degradation

Provide fallback responses when APIs are unavailable.

Write testable code

Use dependency injection and mocking for unit tests.

Error Handling

Comprehensive error handling strategies

Performance

Async operations and optimization techniques

Configuration

Production-ready configuration patterns

Type Safety

Leveraging types for safer code

Getting Started

Gateway API

Agent API

Search API

Embeddings API

Perplexity SDK

Admin & Management

Resources

Legacy API

Overview

Security Best Practices

Environment Variables

API Key Rotation

Rate Limiting and Efficiency

Intelligent Rate Limiting

Request Batching

Production Configuration

Configuration Management

Monitoring and Logging

Error Handling Best Practices

Graceful Degradation

Testing Best Practices

Unit Testing with Mocking

Performance Best Practices Summary

Error Handling

Performance

Configuration

Type Safety

​Overview

​Security Best Practices

​Environment Variables

​API Key Rotation

​Rate Limiting and Efficiency

​Intelligent Rate Limiting

​Request Batching

​Production Configuration

​Configuration Management

​Monitoring and Logging

​Error Handling Best Practices

​Graceful Degradation

​Testing Best Practices

​Unit Testing with Mocking

​Performance Best Practices Summary

​Related Resources

Error Handling

Performance

Configuration

Type Safety

Overview

Security Best Practices

Environment Variables

API Key Rotation

Rate Limiting and Efficiency

Intelligent Rate Limiting

Request Batching

Production Configuration

Configuration Management

Monitoring and Logging

Error Handling Best Practices

Graceful Degradation

Testing Best Practices

Unit Testing with Mocking

Performance Best Practices Summary

Related Resources