Python

#!/usr/bin/env python3

import asyncio
import logging
import os
import time
from datetime import datetime
from typing import Dict, List, Optional, Union, Any
from dataclasses import dataclass, field
import json

import aiohttp
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


@dataclass
class SearchOptions:
    field: str
    value: str
    option: str = 'exact'
    row_limit: int = 1000
    auto_export: Optional[Dict[str, Any]] = None


@dataclass
class SearchResult:
    data: List[Dict[str, Any]]
    total_results: int
    cost: float
    request_id: str
    processing_time: int
    analysis: Optional[Dict[str, Any]] = field(default_factory=dict)


@dataclass
class ApiError(Exception):
    error: str
    message: str
    status_code: int
    retry_after: Optional[int] = None


class LeakZeroClient:

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.leakzero.io/api/v1",
        max_retries: int = 3,
        timeout: int = 120,
        log_level: str = "INFO"
    ):
        if not api_key:
            raise ValueError("API key is required")

        self.api_key = api_key
        self.base_url = base_url.rstrip('/')
        self.max_retries = max_retries
        self.timeout = timeout

        logging.basicConfig(level=getattr(logging, log_level.upper()))
        self.logger = logging.getLogger(__name__)

        self.session = self._setup_session()

        self._async_session = None

    def _setup_session(self) -> requests.Session:
        session = requests.Session()

        retry_strategy = Retry(
            total=self.max_retries,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)

        return session

    def _get_headers(self) -> Dict[str, str]:
        return {
            'x-api-key': self.api_key,
            'x-timestamp': datetime.utcnow().isoformat() + 'Z',
            'x-request-id': self._generate_request_id(),
            'Content-Type': 'application/json',
            'User-Agent': 'LeakZero-Python-SDK/1.0.0'
        }

    def _generate_request_id(self) -> str:
        import uuid
        return f"req_{int(time.time())}_{str(uuid.uuid4())[:8]}"

    def _format_error(self, response: requests.Response) -> ApiError:
        try:
            data = response.json()
            error = data.get('error', 'API_ERROR')
            message = data.get('message', f'HTTP {response.status_code}')
        except:
            error = 'HTTP_ERROR'
            message = f'HTTP {response.status_code}: {response.reason}'

        return ApiError(
            error=error,
            message=message,
            status_code=response.status_code,
            retry_after=response.headers.get('Retry-After')
        )

    def _make_request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict] = None
    ) -> requests.Response:
        url = f"{self.base_url}{endpoint}"
        headers = self._get_headers()

        try:
            if method.upper() == 'POST' and data:
                response = self.session.post(
                    url,
                    json=data,
                    headers=headers,
                    timeout=self.timeout
                )
            else:
                response = self.session.get(
                    url,
                    headers=headers,
                    timeout=self.timeout
                )

            self.logger.info(f"API Request successful: {method.upper()} {endpoint}")

            if response.status_code == 429:
                retry_after = int(response.headers.get('Retry-After', 5))
                self.logger.warning(f"Rate limited. Waiting {retry_after}s")
                time.sleep(min(retry_after, 30))
                return self._make_request(method, endpoint, data)

            response.raise_for_status()
            return response

        except requests.RequestException as e:
            if hasattr(e, 'response') and e.response is not None:
                raise self._format_error(e.response)
            raise ApiError('REQUEST_ERROR', str(e), 0)

    def get_balance(self) -> Dict[str, Any]:
        try:
            response = self._make_request('GET', '/public/balance')
            return response.json()
        except Exception as e:
            self.logger.error(f"Failed to get balance: {e}")
            raise

    def regular_search(self, options: Union[SearchOptions, Dict]) -> SearchResult:
        if isinstance(options, dict):
            options = SearchOptions(**options)

        try:
            payload = {
                'field': options.field,
                'value': options.value,
                'option': options.option,
                'rowLimit': min(options.row_limit, 10000)
            }

            if options.auto_export:
                payload['autoExport'] = json.dumps(options.auto_export)

            self.logger.info(f"Searching for {options.field}: {options.value}")

            response = self._make_request('POST', '/public/search/regular', payload)

            result = SearchResult(
                data=response.json().get('data', []),
                total_results=int(response.headers.get('x-rows-returned', 0)),
                cost=int(response.headers.get('x-charge', 0)) / 1_000_000_000,
                request_id=response.headers.get('x-request-id', 'unknown'),
                processing_time=int(response.headers.get('x-processing-time', 0))
            )

            self.logger.info(f"Found {result.total_results} records (Cost: {result.cost:.6f})")
            return result

        except Exception as e:
            self.logger.error(f"Regular search failed: {e}")
            raise


if __name__ == "__main__":
    api_key = os.getenv('LEAKZERO_API_KEY')
    client = LeakZeroClient(api_key)

    options = SearchOptions(field='email', value='[email protected]')
    result = client.regular_search(options)
    print(f"Found {result.total_results} records")