pip install requests

pip install httpx

import requests
import hashlib
import hmac
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum

class RequestStatus(Enum):
    QUEUED = "QUEUED"
    PROCESSING = "PROCESSING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    PARTIAL = "PARTIAL"

@dataclass
class TxnCheckConfig:
    api_key: str
    base_url: str = "https://api.txncheck.in/api/v1"
    secret_key: Optional[str] = None  # For request signing
    timeout: int = 30

class TxnCheckClient:
    """TxnCheck API Client for Python"""
    
    def __init__(self, config: TxnCheckConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": config.api_key,
            "Content-Type": "application/json"
        })
    
    def _sign_request(self, method: str, path: str, body: str) -> Dict[str, str]:
        """Generate HMAC-SHA256 signature for request"""
        if not self.config.secret_key:
            return {}
        
        timestamp = str(int(time.time() * 1000))
        message = f"{timestamp}.{method}.{path}.{body}"
        signature = hmac.new(
            self.config.secret_key.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        
        return {
            "X-Timestamp": timestamp,
            "X-Signature": signature
        }
    
    def _request(
        self, 
        method: str, 
        endpoint: str, 
        data: Optional[Dict] = None,
        sync_mode: bool = False
    ) -> Dict[str, Any]:
        """Make API request with optional signing"""
        import json
        
        url = f"{self.config.base_url}{endpoint}"
        body = json.dumps(data) if data else ""
        
        # Add async flag
        if data and sync_mode:
            data["async"] = False
        
        # Sign request if secret key is configured
        headers = self._sign_request(method.upper(), endpoint, body)
        
        response = self.session.request(
            method=method,
            url=url,
            json=data,
            headers=headers,
            timeout=self.config.timeout
        )
        
        response.raise_for_status()
        return response.json()
    
    # ==================
    # Verification Methods
    # ==================
    
    def upi_by_mobile(
        self, 
        mobile: str, 
        sync: bool = False
    ) -> Dict[str, Any]:
        """
        Get UPI VPAs linked to a mobile number.
        
        Args:
            mobile: Mobile number in +91XXXXXXXXXX format
            sync: If True, wait for result (up to 30s)
        
        Returns:
            Request ID and status (async) or full result (sync)
        """
        return self._request(
            "POST", 
            "/upi-by-mobile", 
            {"mobile": mobile},
            sync_mode=sync
        )
    
    def kyc_by_mobile(
        self, 
        mobile: str, 
        sync: bool = False
    ) -> Dict[str, Any]:
        """
        Get KYC data by mobile number.
        
        Args:
            mobile: Mobile number in +91XXXXXXXXXX format
            sync: If True, wait for result (up to 30s)
        
        Returns:
            Request ID and status (async) or full result (sync)
        """
        return self._request(
            "POST", 
            "/kyc-by-mobile", 
            {"mobile": mobile},
            sync_mode=sync
        )
    
    def vpa_chargeback_check(
        self, 
        vpas: List[str], 
        sync: bool = False
    ) -> Dict[str, Any]:
        """
        Check VPAs against blocklist.
        
        Args:
            vpas: List of VPA addresses (max 100)
            sync: If True, wait for result (up to 30s)
        
        Returns:
            Request ID and status (async) or full result (sync)
        """
        return self._request(
            "POST", 
            "/vpa-chargeback-check", 
            {"vpas": vpas},
            sync_mode=sync
        )
    
    def full_check(
        self, 
        mobile: str, 
        sync: bool = False
    ) -> Dict[str, Any]:
        """
        Full verification: UPI + KYC + Chargeback check.
        
        Args:
            mobile: Mobile number in +91XXXXXXXXXX format
            sync: If True, wait for result (up to 30s)
        
        Returns:
            Request ID and status (async) or full result (sync)
        """
        return self._request(
            "POST", 
            "/full-check", 
            {"mobile": mobile},
            sync_mode=sync
        )
    
    def bulk_vpa_check(
        self, 
        vpas: List[str], 
        sync: bool = False
    ) -> Dict[str, Any]:
        """
        Bulk check VPAs against blocklist.
        
        Args:
            vpas: List of VPA addresses (1-100)
            sync: If True, wait for result (up to 30s)
        
        Returns:
            Request ID and status (async) or full result (sync)
        """
        return self._request(
            "POST", 
            "/bulk/vpa-chargeback-check", 
            {"vpas": vpas},
            sync_mode=sync
        )
    
    def get_request_status(self, request_id: str) -> Dict[str, Any]:
        """
        Get status and results of a request.
        
        Args:
            request_id: UUID of the request
        
        Returns:
            Request status and result data
        """
        return self._request("GET", f"/requests/{request_id}")
    
    def wait_for_result(
        self, 
        request_id: str, 
        poll_interval: float = 2.0,
        max_wait: float = 60.0
    ) -> Dict[str, Any]:
        """
        Poll for request completion.
        
        Args:
            request_id: UUID of the request
            poll_interval: Seconds between polls
            max_wait: Maximum seconds to wait
        
        Returns:
            Completed request data
        
        Raises:
            TimeoutError: If max_wait exceeded
        """
        terminal_statuses = {
            RequestStatus.COMPLETED.value,
            RequestStatus.FAILED.value,
            RequestStatus.PARTIAL.value
        }
        
        start_time = time.time()
        while time.time() - start_time < max_wait:
            result = self.get_request_status(request_id)
            if result.get("status") in terminal_statuses:
                return result
            time.sleep(poll_interval)
        
        raise TimeoutError(f"Request {request_id} did not complete within {max_wait}s")

from txncheck import TxnCheckClient, TxnCheckConfig

# Initialize client
config = TxnCheckConfig(api_key="fb_your_api_key_here")
client = TxnCheckClient(config)

# Submit request
response = client.upi_by_mobile("+919876543210")
print(f"Request ID: {response['requestId']}")
# Output: Request ID: 550e8400-e29b-41d4-a716-446655440000

# Poll for result
result = client.wait_for_result(response['requestId'])
print(f"Status: {result['status']}")
print(f"UPI addresses: {result['result']['upi']}")

# Get result immediately (waits up to 30 seconds)
result = client.upi_by_mobile("+919876543210", sync=True)
print(f"Status: {result['status']}")
print(f"Name: {result['result']['name']}")
print(f"UPI addresses: {result['result']['upi']}")

# Comprehensive verification
result = client.full_check("+919876543210", sync=True)

if result['status'] == 'COMPLETED':
    # UPI data
    upi_data = result['result']['upiByMobile']
    print(f"Name: {upi_data['name']}")
    print(f"VPAs: {upi_data['upi']}")
    
    # KYC data
    kyc_data = result['result']['kycByMobile']
    print(f"PAN: {kyc_data['pan']}")
    print(f"DOB: {kyc_data['dob']}")
    
    # Blocklist status
    blocklist = result['result']['vpaChargebackCheck']
    print(f"Blocklisted VPAs: {len(blocklist['blocklisted'])}")

# Check multiple VPAs
vpas = ["user1@upi", "user2@paytm", "suspicious@ybl"]
result = client.vpa_chargeback_check(vpas, sync=True)

for vpa in result['result']['blocklisted']:
    print(f"⚠️ BLOCKED: {vpa['vpa']}")

for vpa in result['result']['clean']:
    print(f"✓ Clean: {vpa['vpa']}")

import requests

class TxnCheckError(Exception):
    """Base exception for TxnCheck errors"""
    def __init__(self, status_code: int, message: str, error: str):
        self.status_code = status_code
        self.message = message
        self.error = error
        super().__init__(f"{status_code} {error}: {message}")

def handle_api_error(response: requests.Response):
    """Convert HTTP errors to TxnCheckError"""
    try:
        data = response.json()
        raise TxnCheckError(
            status_code=response.status_code,
            message=data.get('message', 'Unknown error'),
            error=data.get('error', 'Error')
        )
    except ValueError:
        raise TxnCheckError(
            status_code=response.status_code,
            message=response.text,
            error='Unknown Error'
        )

# Usage with retry logic
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(requests.exceptions.RequestException)
)
def verify_with_retry(client: TxnCheckClient, mobile: str):
    try:
        return client.upi_by_mobile(mobile, sync=True)
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:  # Rate limited
            raise  # Will retry
        if e.response.status_code >= 500:  # Server error
            raise  # Will retry
        handle_api_error(e.response)  # Client error - don't retry

import hmac
import hashlib

def verify_webhook_signature(
    payload: bytes,
    signature: str,
    timestamp: str,
    secret_key: str,
    max_age_seconds: int = 300
) -> bool:
    """
    Verify webhook signature from TxnCheck.
    
    Args:
        payload: Raw request body bytes
        signature: X-Webhook-Signature header value
        timestamp: X-Webhook-Timestamp header value
        secret_key: Your webhook secret key
        max_age_seconds: Maximum age of webhook to accept
    
    Returns:
        True if signature is valid
    """
    import time
    
    # Check timestamp freshness
    webhook_time = int(timestamp)
    current_time = int(time.time() * 1000)
    age_ms = current_time - webhook_time
    
    if age_ms > max_age_seconds * 1000:
        return False  # Webhook too old
    
    # Compute expected signature
    message = f"{timestamp}.{payload.decode('utf-8')}"
    expected = hmac.new(
        secret_key.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    
    # Constant-time comparison
    return hmac.compare_digest(signature, expected)

# Flask example
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route('/webhook/fraud-buster', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Webhook-Signature')
    timestamp = request.headers.get('X-Webhook-Timestamp')
    
    if not verify_webhook_signature(
        request.data, signature, timestamp, WEBHOOK_SECRET
    ):
        return jsonify({"error": "Invalid signature"}), 401
    
    data = request.json
    event = data['event']
    
    if event == 'request.completed':
        handle_completed(data['data'])
    elif event == 'request.failed':
        handle_failed(data['data'])
    
    return jsonify({"status": "ok"}), 200

import httpx
import asyncio
from typing import Optional, Dict, Any, List

class AsyncTxnCheckClient:
    """Async TxnCheck API Client using httpx"""
    
    def __init__(self, config: TxnCheckConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=config.base_url,
            headers={
                "X-API-Key": config.api_key,
                "Content-Type": "application/json"
            },
            timeout=config.timeout
        )
    
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, *args):
        await self.client.aclose()
    
    async def upi_by_mobile(
        self, 
        mobile: str, 
        sync: bool = False
    ) -> Dict[str, Any]:
        data = {"mobile": mobile}
        if sync:
            data["async"] = False
        
        response = await self.client.post("/upi-by-mobile", json=data)
        response.raise_for_status()
        return response.json()
    
    async def verify_multiple(
        self, 
        mobiles: List[str]
    ) -> List[Dict[str, Any]]:
        """Verify multiple mobile numbers concurrently"""
        tasks = [self.upi_by_mobile(m, sync=True) for m in mobiles]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Usage
async def main():
    config = TxnCheckConfig(api_key="fb_your_api_key_here")
    
    async with AsyncTxnCheckClient(config) as client:
        # Single request
        result = await client.upi_by_mobile("+919876543210", sync=True)
        print(result)
        
        # Concurrent requests
        mobiles = ["+919876543210", "+919876543211", "+919876543212"]
        results = await client.verify_multiple(mobiles)
        for r in results:
            if isinstance(r, Exception):
                print(f"Error: {r}")
            else:
                print(f"Verified: {r['result']['name']}")

asyncio.run(main())

"""
Complete example: E-commerce checkout verification
"""
from txncheck import TxnCheckClient, TxnCheckConfig
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def verify_customer_for_checkout(
    client: TxnCheckClient,
    mobile: str,
    payment_vpa: str
) -> dict:
    """
    Verify customer before processing payment.
    
    Returns:
        dict with verification results and risk score
    """
    result = {
        "verified": False,
        "risk_score": 0,
        "details": {}
    }
    
    try:
        # Step 1: Full verification
        logger.info(f"Starting verification for {mobile}")
        verification = client.full_check(mobile, sync=True)
        
        if verification['status'] != 'COMPLETED':
            logger.warning(f"Verification incomplete: {verification['status']}")
            result["risk_score"] = 100
            return result
        
        # Step 2: Extract data
        upi_data = verification['result'].get('upiByMobile', {})
        kyc_data = verification['result'].get('kycByMobile', {})
        blocklist = verification['result'].get('vpaChargebackCheck', {})
        
        # Step 3: Check if payment VPA belongs to user
        user_vpas = upi_data.get('upi', [])
        vpa_belongs_to_user = payment_vpa.lower() in [v.lower() for v in user_vpas]
        
        if not vpa_belongs_to_user:
            logger.warning(f"Payment VPA {payment_vpa} not linked to {mobile}")
            result["risk_score"] += 50
        
        # Step 4: Check blocklist
        blocklisted_vpas = [v['vpa'] for v in blocklist.get('blocklisted', [])]
        if payment_vpa.lower() in [v.lower() for v in blocklisted_vpas]:
            logger.error(f"Payment VPA {payment_vpa} is BLOCKLISTED!")
            result["risk_score"] = 100
            result["details"]["blocklisted"] = True
            return result
        
        # Step 5: Build result
        result["verified"] = result["risk_score"] < 50
        result["details"] = {
            "name": kyc_data.get('fullName', upi_data.get('name')),
            "pan_verified": bool(kyc_data.get('pan')),
            "vpa_linked": vpa_belongs_to_user,
            "vpa_count": len(user_vpas),
            "blocklisted": False
        }
        
        logger.info(f"Verification complete. Risk score: {result['risk_score']}")
        return result
        
    except Exception as e:
        logger.error(f"Verification failed: {e}")
        result["risk_score"] = 100
        result["error"] = str(e)
        return result

# Usage
if __name__ == "__main__":
    config = TxnCheckConfig(api_key="fb_your_api_key_here")
    client = TxnCheckClient(config)
    
    verification = verify_customer_for_checkout(
        client,
        mobile="+919876543210",
        payment_vpa="user@upi"
    )
    
    if verification["verified"]:
        print("✓ Customer verified, proceed with payment")
    else:
        print(f"⚠️ Verification failed. Risk score: {verification['risk_score']}")