pan_aadhaar_link.py•3.52 kB
"""PAN-Aadhaar link status tool implementation."""
import hashlib
import time
from typing import Any, Dict
from pydantic import ValidationError as PydanticValidationError
from src.clients.kyc_api_client import KYCAPIClient
from src.models.requests import PANAadhaarLinkRequest
from src.models.responses import PANAadhaarLinkResponse
from src.tools.base_tool import BaseTool
from src.utils.logger import get_logger
logger = get_logger(__name__)
class PANAadhaarLinkTool(BaseTool):
"""
Tool for checking PAN-Aadhaar link status.
"""
def __init__(self, api_client: KYCAPIClient):
"""
Initialize PAN-Aadhaar link tool.
Args:
api_client: KYC API client instance
"""
self.api_client = api_client
def get_name(self) -> str:
"""Return tool name."""
return "check_pan_aadhaar_link"
async def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Check if PAN and Aadhaar are linked.
Args:
params: Dictionary containing:
- pan: Individual PAN number (4th character must be 'P')
- aadhaar_number: 12-digit Aadhaar number
- consent: User consent ('Y' or 'y')
- reason: Reason for checking
Returns:
Link status with descriptive message
"""
logger.info("pan_aadhaar_link_check_started", pan=params.get("pan"))
try:
# Validate input using Pydantic model
request = PANAadhaarLinkRequest(**params)
# Prepare API request payload
api_payload = {
"@entity": "in.co.sandbox.kyc.pan_aadhaar.status",
"pan": request.pan,
"aadhaar_number": request.aadhaar_number,
"consent": request.consent,
"reason": request.reason,
}
# Call KYC API
response = await self.api_client.post(
endpoint="/kyc/pan-aadhaar/status", data=api_payload
)
# Parse and validate response
data = response.get("data", {})
result = PANAadhaarLinkResponse(**data)
# Format response
output = {
"linked": result.aadhaar_seeding_status == "y",
"status": result.aadhaar_seeding_status,
"message": result.message,
"checked_at": int(time.time()),
}
logger.info(
"pan_aadhaar_link_check_completed",
pan=request.pan,
linked=output["linked"],
)
return output
except PydanticValidationError as e:
logger.error("pan_aadhaar_link_validation_error", error=str(e))
raise ValueError(f"Invalid input parameters: {str(e)}")
except Exception as e:
logger.error("pan_aadhaar_link_check_failed", error=str(e), exc_info=True)
raise
def get_cache_key(self, params: Dict[str, Any]) -> str:
"""
Generate cache key from PAN and Aadhaar.
Args:
params: Input parameters
Returns:
Cache key
"""
key_data = f"{params.get('pan')}|{params.get('aadhaar_number')}"
hash_key = hashlib.sha256(key_data.encode()).hexdigest()
return f"pan_aadhaar:{hash_key}"
def get_cache_ttl(self) -> int:
"""Get cache TTL for PAN-Aadhaar link check."""
return 7200 # 2 hours