vickytechkey
c6b177ac55
All checks were successful
Deploy Beta (NATIVE) / deploy (push) Successful in 31s
120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
import time
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class DocumentType(str, Enum):
|
|
PAN = "pan"
|
|
AADHAAR = "aadhaar"
|
|
DRIVING_LICENSE = "driving_license"
|
|
VOTER_ID = "voter_id"
|
|
PASSPORT = "passport"
|
|
BANK_ACCOUNT = "bank_account"
|
|
GSTIN = "gstin"
|
|
CUSTOM = "custom"
|
|
|
|
|
|
class VerificationStatus(str, Enum):
|
|
SUCCESS = "SUCCESS"
|
|
FAILED = "FAILED"
|
|
ERROR = "ERROR"
|
|
SKIPPED = "SKIPPED"
|
|
|
|
|
|
@dataclass
|
|
class VendorResult:
|
|
"""Standardized result returned by every vendor verification adapter."""
|
|
vendor_name: str
|
|
is_success: bool
|
|
status: VerificationStatus
|
|
data: Dict[str, Any] = field(default_factory=dict)
|
|
raw_response: Any = None
|
|
error_code: Optional[str] = None
|
|
error_message: Optional[str] = None
|
|
latency_ms: float = 0.0
|
|
retryable: bool = False
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"vendor_name": self.vendor_name,
|
|
"is_success": self.is_success,
|
|
"status": self.status.value if isinstance(self.status, VerificationStatus) else str(self.status),
|
|
"data": self.data,
|
|
"raw_response": self.raw_response,
|
|
"error_code": self.error_code,
|
|
"error_message": self.error_message,
|
|
"latency_ms": round(self.latency_ms, 2),
|
|
"retryable": self.retryable,
|
|
}
|
|
|
|
|
|
class BaseVendor:
|
|
"""
|
|
Abstract Base Class for all Verification Vendors (e.g. Codeplus, Cashfree, Karza, Surepass).
|
|
Subclasses implement vendor-specific logic while adhering to this unified contract.
|
|
"""
|
|
name: str = "base_vendor"
|
|
display_name: str = "Base Vendor"
|
|
priority: int = 100
|
|
is_active: bool = True
|
|
supported_documents: List[str] = []
|
|
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None):
|
|
self.config = config or {}
|
|
|
|
def supports(self, document_type: str) -> bool:
|
|
"""Check if this vendor supports the requested document type."""
|
|
doc_key = document_type.lower()
|
|
return doc_key in [d.lower() for d in self.supported_documents] or "all" in self.supported_documents
|
|
|
|
def verify(
|
|
self,
|
|
document_type: str,
|
|
data: Dict[str, Any],
|
|
files: Optional[Dict[str, Any]] = None,
|
|
) -> VendorResult:
|
|
"""
|
|
Executes verification with latency tracking and universal exception handling.
|
|
Delegates actual execution to `execute_verification`.
|
|
"""
|
|
start_time = time.time()
|
|
doc_type_clean = document_type.lower()
|
|
|
|
if not self.supports(doc_type_clean):
|
|
latency = (time.time() - start_time) * 1000
|
|
return VendorResult(
|
|
vendor_name=self.name,
|
|
is_success=False,
|
|
status=VerificationStatus.SKIPPED,
|
|
error_code="DOC_NOT_SUPPORTED",
|
|
error_message=f"Vendor '{self.name}' does not support document type '{document_type}'",
|
|
latency_ms=latency,
|
|
retryable=True,
|
|
)
|
|
|
|
try:
|
|
result = self.execute_verification(doc_type_clean, data, files)
|
|
result.latency_ms = (time.time() - start_time) * 1000
|
|
result.vendor_name = self.name
|
|
return result
|
|
except Exception as exc:
|
|
latency = (time.time() - start_time) * 1000
|
|
return VendorResult(
|
|
vendor_name=self.name,
|
|
is_success=False,
|
|
status=VerificationStatus.ERROR,
|
|
error_code=exc.__class__.__name__,
|
|
error_message=str(exc),
|
|
latency_ms=latency,
|
|
retryable=True,
|
|
)
|
|
|
|
def execute_verification(
|
|
self,
|
|
document_type: str,
|
|
data: Dict[str, Any],
|
|
files: Optional[Dict[str, Any]] = None,
|
|
) -> VendorResult:
|
|
"""Must be implemented by subclasses."""
|
|
raise NotImplementedError("Subclasses must implement execute_verification()")
|