940 lines
35 KiB
Python
940 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FastAPI Equifax Provider
|
|
Provides REST API endpoints for Equifax transaction services
|
|
Based on the working example in main.py
|
|
"""
|
|
|
|
from multiprocessing import process
|
|
import os
|
|
import json
|
|
import time
|
|
import asyncio
|
|
import logging
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional, List
|
|
from datetime import datetime, timedelta
|
|
|
|
import requests
|
|
from fastapi import FastAPI, HTTPException, Depends, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field, validator
|
|
from dotenv import load_dotenv
|
|
from pymongo import MongoClient
|
|
from pymongo.errors import ConnectionFailure
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
print("MONGODB_URI:", os.getenv("MONGODB_URI"))
|
|
|
|
# MongoDB connection setup
|
|
MONGODB_URI = os.getenv("MONGODB_URI", "mongodb://localhost:27017/duxiter")
|
|
CACHE_TTL_SECONDS = int(os.getenv("EQUIFAX_CACHE_TTL_SECONDS", "604800")) # Default 1 week
|
|
|
|
try:
|
|
mongo_client = MongoClient(MONGODB_URI)
|
|
# Test connection
|
|
mongo_client.admin.command('ping')
|
|
db = mongo_client.get_database()
|
|
cache_collection = db.equifax_cache
|
|
|
|
# Create TTL index for automatic document expiration
|
|
cache_collection.create_index("expires_at", expireAfterSeconds=0)
|
|
logger.info(f"MongoDB connected successfully. Cache TTL: {CACHE_TTL_SECONDS} seconds")
|
|
except ConnectionFailure as e:
|
|
logger.error(f"MongoDB connection failed: {e}")
|
|
mongo_client = None
|
|
db = None
|
|
cache_collection = None
|
|
|
|
# Constants
|
|
TOKEN_CACHE_FILE = Path(".token_cache.json")
|
|
EQUIFAX_RESPONSES_DIR = Path(__file__).resolve().parents[2] / "equifax_responses"
|
|
|
|
ENDPOINTS = {
|
|
"uat": {
|
|
"token": "https://api.uat.latam.equifax.com/v2/oauth/token",
|
|
"execute": "https://api.uat.latam.equifax.com/datos-comerciales/transaction/execute",
|
|
},
|
|
"prod": {
|
|
"token": "https://api.latam.equifax.com/v2/oauth/token",
|
|
"execute": "https://api.latam.equifax.com/datos-comerciales/transaction/execute",
|
|
},
|
|
}
|
|
|
|
# Pydantic Models
|
|
class PersonalInformation(BaseModel):
|
|
chileanRut: str = Field(..., description="Chilean RUT number")
|
|
# chileanSerialNumber: Optional[str] = Field(None, description="Chilean serial number (not required for empresarial)")
|
|
|
|
@validator('chileanRut')
|
|
def validate_rut(cls, v):
|
|
if not v or not v.strip():
|
|
raise ValueError('Chilean RUT is required')
|
|
return v.strip()
|
|
|
|
class PrimaryConsumer(BaseModel):
|
|
personalInformation: PersonalInformation
|
|
|
|
class Applicants(BaseModel):
|
|
primaryConsumer: PrimaryConsumer
|
|
|
|
class ProductData(BaseModel):
|
|
billTo: str = Field(..., description="Bill to information")
|
|
shipTo: str = Field(..., description="Ship to information")
|
|
productName: str = Field(..., description="Product name")
|
|
productOrch: str = Field(..., description="Product orchestration")
|
|
configuration: str = Field(default="Config", description="Configuration")
|
|
customer: str = Field(..., description="Customer information")
|
|
model: str = Field(..., description="Model information")
|
|
|
|
@validator('*', pre=True)
|
|
def strip_strings(cls, v):
|
|
if isinstance(v, str):
|
|
return v.strip()
|
|
return v
|
|
|
|
class EquifaxRequest(BaseModel):
|
|
applicants: Applicants
|
|
productData: ProductData
|
|
requestType: Optional[str] = Field(default="personal", description="Request type: personal or empresarial")
|
|
|
|
|
|
|
|
|
|
|
|
class RutOnlyTransactionRequest(BaseModel):
|
|
"""Simplified request for RUT-only transactions - only requires RUT and request type"""
|
|
chileanRut: str = Field(..., description="Chilean RUT number", example="77827780-8")
|
|
requestType: Optional[str] = Field(default="personal", description="Request type: personal or empresarial", example="personal")
|
|
|
|
class Config:
|
|
schema_extra = {
|
|
"example": {
|
|
"chileanRut": "77827780-8",
|
|
"requestType": "personal"
|
|
}
|
|
}
|
|
|
|
class TokenResponse(BaseModel):
|
|
access_token: str
|
|
expires_in: int
|
|
cached: bool
|
|
|
|
class HealthResponse(BaseModel):
|
|
status: str
|
|
timestamp: str
|
|
version: str
|
|
|
|
class ErrorResponse(BaseModel):
|
|
error: str
|
|
detail: str
|
|
timestamp: str
|
|
|
|
# FastAPI App
|
|
app = FastAPI(
|
|
title="Equifax Provider API",
|
|
description="REST API for Equifax transaction services",
|
|
version="1.0.0",
|
|
docs_url="/docs",
|
|
redoc_url="/redoc"
|
|
)
|
|
|
|
# CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Configure appropriately for production
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Cache helper functions
|
|
def generate_cache_key(env_key: str, payload: Dict[str, Any]) -> str:
|
|
"""Generate a unique cache key based on environment and payload"""
|
|
# Create a normalized string representation of the payload
|
|
payload_str = json.dumps(payload, sort_keys=True, separators=(',', ':'))
|
|
# Create hash of the payload for consistent key generation
|
|
payload_hash = hashlib.sha256(payload_str.encode()).hexdigest()
|
|
return f"equifax_cache:{env_key}:{payload_hash}"
|
|
|
|
def get_cached_transaction(cache_key: str) -> Optional[Dict[str, Any]]:
|
|
"""Retrieve cached transaction result if available and not expired"""
|
|
if cache_collection is None:
|
|
return None
|
|
|
|
try:
|
|
cached_doc = cache_collection.find_one({"_id": cache_key})
|
|
if cached_doc:
|
|
logger.info(f"Cache hit for key: {cache_key}")
|
|
return cached_doc.get("result")
|
|
else:
|
|
logger.info(f"Cache miss for key: {cache_key}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error retrieving from cache: {e}")
|
|
return None
|
|
|
|
def save_transaction_to_cache(cache_key: str, result: Dict[str, Any]) -> None:
|
|
"""Save transaction result to cache with TTL"""
|
|
if cache_collection is None:
|
|
return
|
|
|
|
try:
|
|
expires_at = datetime.utcnow() + timedelta(seconds=CACHE_TTL_SECONDS)
|
|
cache_doc = {
|
|
"_id": cache_key,
|
|
"result": result,
|
|
"cached_at": datetime.utcnow(),
|
|
"expires_at": expires_at
|
|
}
|
|
|
|
# Use upsert to replace existing cache entry
|
|
cache_collection.replace_one(
|
|
{"_id": cache_key},
|
|
cache_doc,
|
|
upsert=True
|
|
)
|
|
logger.info(f"Transaction result cached with key: {cache_key}, expires at: {expires_at}")
|
|
except Exception as e:
|
|
logger.error(f"Error saving to cache: {e}")
|
|
|
|
def clear_transaction_cache() -> int:
|
|
"""Clear all cached transactions and return count of deleted documents"""
|
|
if cache_collection is None:
|
|
return 0
|
|
|
|
try:
|
|
result = cache_collection.delete_many({})
|
|
logger.info(f"Cleared {result.deleted_count} cached transactions")
|
|
return result.deleted_count
|
|
except Exception as e:
|
|
logger.error(f"Error clearing cache: {e}")
|
|
return 0
|
|
|
|
def get_cache_stats() -> Dict[str, Any]:
|
|
"""Get cache statistics"""
|
|
if cache_collection is None:
|
|
return {"error": "Cache not available"}
|
|
|
|
try:
|
|
total_docs = cache_collection.count_documents({})
|
|
expired_docs = cache_collection.count_documents({
|
|
"expires_at": {"$lt": datetime.utcnow()}
|
|
})
|
|
|
|
return {
|
|
"total_cached_transactions": total_docs,
|
|
"expired_transactions": expired_docs,
|
|
"active_transactions": total_docs - expired_docs,
|
|
"cache_ttl_seconds": CACHE_TTL_SECONDS,
|
|
"mongodb_connected": True
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error getting cache stats: {e}")
|
|
return {"error": str(e), "mongodb_connected": False}
|
|
|
|
# Utility Functions (from main.py)
|
|
def now() -> float:
|
|
return time.time()
|
|
|
|
def read_cache() -> Optional[dict]:
|
|
"""Read token cache from file"""
|
|
if TOKEN_CACHE_FILE.exists():
|
|
try:
|
|
logger.debug(f"Reading token cache from {TOKEN_CACHE_FILE.absolute()}")
|
|
with TOKEN_CACHE_FILE.open("r", encoding="utf-8") as f:
|
|
cache_data = json.load(f)
|
|
logger.debug(f"Cache contains {len(cache_data)} entries")
|
|
return cache_data
|
|
except Exception as e:
|
|
logger.warning(f"Failed to read token cache: {e}")
|
|
return None
|
|
else:
|
|
logger.debug("No token cache file found")
|
|
return None
|
|
|
|
def write_cache(data: dict) -> None:
|
|
"""Write token cache to file"""
|
|
try:
|
|
logger.debug(f"Writing token cache to {TOKEN_CACHE_FILE.absolute()}")
|
|
with TOKEN_CACHE_FILE.open("w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
logger.debug("Token cache written successfully")
|
|
except Exception as e:
|
|
logger.error(f"Failed to write token cache: {e}")
|
|
|
|
def get_cached_token(env_key: str) -> Optional[str]:
|
|
"""Get cached token if valid"""
|
|
logger.debug(f"Checking cached token for environment: {env_key}")
|
|
cache = read_cache()
|
|
if not cache:
|
|
logger.debug("No cache available")
|
|
return None
|
|
|
|
item = cache.get(env_key)
|
|
if not item:
|
|
logger.debug(f"No cached token found for environment: {env_key}")
|
|
return None
|
|
|
|
access_token = item.get("access_token")
|
|
expires_at = item.get("expires_at", 0)
|
|
|
|
if not access_token:
|
|
logger.debug("Cached entry has no access_token")
|
|
return None
|
|
|
|
current_time = now()
|
|
time_until_expiry = expires_at - current_time
|
|
|
|
logger.debug(f"Token expires in {time_until_expiry:.1f} seconds")
|
|
|
|
# Add a small safety buffer of 60s
|
|
if current_time < (expires_at - 60):
|
|
logger.info(f"Using cached token for {env_key} (expires in {time_until_expiry:.1f}s)")
|
|
return access_token
|
|
else:
|
|
logger.debug(f"Cached token expired or expires soon (in {time_until_expiry:.1f}s)")
|
|
return None
|
|
|
|
def save_cached_token(env_key: str, access_token: str, expires_in_secs: int) -> None:
|
|
"""Save token to cache"""
|
|
logger.debug(f"Saving token for {env_key} (expires in {expires_in_secs}s)")
|
|
cache = read_cache() or {}
|
|
current_time = now()
|
|
|
|
cache[env_key] = {
|
|
"access_token": access_token,
|
|
"expires_at": current_time + int(expires_in_secs),
|
|
"saved_at": current_time,
|
|
}
|
|
|
|
write_cache(cache)
|
|
logger.info(f"Token cached for {env_key}, expires at {time.ctime(cache[env_key]['expires_at'])}")
|
|
|
|
def fetch_token(env_key: str, client_id: str, client_secret: str, scope: str) -> str:
|
|
"""Fetch OAuth token with detailed logging"""
|
|
logger.info(f"Fetching token for environment: {env_key}")
|
|
|
|
url = ENDPOINTS[env_key]["token"]
|
|
logger.info(f"Token endpoint: {url}")
|
|
|
|
scope = "https://api.latam.equifax.com/datos-comerciales/transaction"
|
|
|
|
# Default OAuth2 client_credentials
|
|
data = {
|
|
"grant_type": "client_credentials",
|
|
"scope": scope,
|
|
}
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
}
|
|
|
|
# Try different authentication methods
|
|
try_methods = [
|
|
("basic", {"auth": (client_id, client_secret), "data": data}),
|
|
("body", {"auth": None, "data": {**data, "client_id": client_id, "client_secret": client_secret}}),
|
|
]
|
|
|
|
last_err = None
|
|
for method_name, params in try_methods:
|
|
try:
|
|
logger.info(f"Trying authentication method: {method_name}")
|
|
resp = requests.post(url, headers=headers, **params, timeout=30)
|
|
logger.info(f"Token response status: {resp.status_code}")
|
|
|
|
if resp.status_code == 200:
|
|
payload = resp.json()
|
|
access_token = payload.get("access_token")
|
|
expires_in = payload.get("expires_in_secs") or payload.get("expires_in") or 300
|
|
if not access_token:
|
|
raise ValueError(f"Token response missing access_token. Payload: {payload}")
|
|
|
|
logger.info(f"Successfully fetched new token (expires in {expires_in}s)")
|
|
save_cached_token(env_key, access_token, int(expires_in))
|
|
return access_token
|
|
else:
|
|
last_err = f"{resp.status_code} {resp.text}"
|
|
logger.debug(f"Method {method_name} failed: {last_err}")
|
|
except Exception as e:
|
|
last_err = str(e)
|
|
logger.debug(f"Method {method_name} failed with exception: {last_err}")
|
|
|
|
logger.error(f"Failed to obtain token from {url}. Last error: {last_err}")
|
|
raise RuntimeError(f"Failed to obtain token from {url}. Last error: {last_err}")
|
|
|
|
def get_token(env_key: str, client_id: str, client_secret: str, scope: str) -> str:
|
|
"""Get token with cache check"""
|
|
logger.debug(f"Getting token for environment: {env_key}")
|
|
|
|
# Check cache first
|
|
token = get_cached_token(env_key)
|
|
if token:
|
|
return token
|
|
|
|
# Fetch new token
|
|
logger.debug("No valid cached token found, fetching new token")
|
|
return fetch_token(env_key, client_id, client_secret, scope)
|
|
|
|
def execute_transaction(env_key: str, token: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Execute transaction with detailed logging, MongoDB caching, and automatic token refresh on 401"""
|
|
# Generate cache key based on environment and payload
|
|
cache_key = generate_cache_key(env_key, payload)
|
|
|
|
# Check cache first
|
|
cached_result = get_cached_transaction(cache_key)
|
|
if cached_result:
|
|
logger.info("Returning cached transaction result")
|
|
# Add cache metadata to response
|
|
cached_result["_cache_info"] = {
|
|
"cached": True,
|
|
"cache_key": cache_key,
|
|
"retrieved_at": datetime.utcnow().isoformat()
|
|
}
|
|
return cached_result
|
|
|
|
# Cache miss - proceed with API call
|
|
return _execute_transaction_with_retry(env_key, token, payload, cache_key)
|
|
|
|
|
|
def _execute_transaction_with_retry(env_key: str, token: str, payload: Dict[str, Any], cache_key: str, is_retry: bool = False) -> Dict[str, Any]:
|
|
"""Internal function to execute transaction with retry logic for 401 errors"""
|
|
url = ENDPOINTS[env_key]["execute"]
|
|
|
|
logger.info(f"Executing transaction on {env_key} environment{'(retry with refreshed token)' if is_retry else ''}")
|
|
logger.info(f"Transaction URL: {url}")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
logger.info(f"Request headers: {headers}")
|
|
if not is_retry: # Only log payload on first attempt to avoid spam
|
|
logger.info(f"Request payload: {json.dumps(payload, indent=2)}")
|
|
|
|
|
|
resp = requests.post(url, headers=headers, json=payload, timeout=60)
|
|
logger.info(f"Transaction response status: {resp.status_code}")
|
|
|
|
# Handle 401 error with token refresh
|
|
if resp.status_code == 401:
|
|
if is_retry:
|
|
# If this is already a retry, don't retry again
|
|
logger.error("401 Unauthorized: Token refresh failed, both original and refreshed tokens are invalid")
|
|
raise HTTPException(status_code=401, detail="Token refresh failed - authentication error persists")
|
|
|
|
logger.warning("401 Unauthorized: Token invalid or expired, attempting to refresh token")
|
|
|
|
# Clear the cached token to force refresh
|
|
_clear_cached_token(env_key)
|
|
|
|
# Get credentials and fetch a new token
|
|
client_id, client_secret = get_credentials()
|
|
scope = "https://api.latam.equifax.com/datos-comerciales/transaction"
|
|
|
|
try:
|
|
new_token = fetch_token(env_key, client_id, client_secret, scope)
|
|
logger.info("Successfully refreshed token, retrying transaction")
|
|
|
|
# Retry with the new token
|
|
return _execute_transaction_with_retry(env_key, new_token, payload, cache_key, is_retry=True)
|
|
except Exception as e:
|
|
logger.error(f"Failed to refresh token: {e}")
|
|
raise HTTPException(status_code=401, detail=f"Token refresh failed: {str(e)}")
|
|
|
|
# Handle other error patterns
|
|
if resp.status_code == 404:
|
|
logger.error("404 Not Found: Requested orchestration not found")
|
|
raise HTTPException(status_code=404, detail="Requested orchestration not found")
|
|
if resp.status_code == 400:
|
|
logger.error(f"400 Bad Request: {resp.text}")
|
|
raise HTTPException(status_code=400, detail=f"Bad Request: {resp.text}")
|
|
if resp.status_code >= 500:
|
|
logger.error(f"{resp.status_code} Server error: {resp.text}")
|
|
raise HTTPException(status_code=502, detail=f"Equifax server error: {resp.status_code}")
|
|
|
|
try:
|
|
result = resp.json()
|
|
logger.info("Transaction completed successfully")
|
|
|
|
# Save successful result to cache
|
|
save_transaction_to_cache(cache_key, result)
|
|
|
|
# Add cache metadata to response
|
|
result["_cache_info"] = {
|
|
"cached": False,
|
|
"cache_key": cache_key,
|
|
"cached_at": datetime.utcnow().isoformat(),
|
|
"token_refreshed": is_retry
|
|
}
|
|
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse JSON response: {e}")
|
|
raw_result = {"raw": resp.text}
|
|
|
|
# Save raw response to cache as well (in case it's a valid response format)
|
|
save_transaction_to_cache(cache_key, raw_result)
|
|
|
|
# Add cache metadata to response
|
|
raw_result["_cache_info"] = {
|
|
"cached": False,
|
|
"cache_key": cache_key,
|
|
"cached_at": datetime.utcnow().isoformat(),
|
|
"token_refreshed": is_retry
|
|
}
|
|
|
|
return raw_result
|
|
|
|
|
|
def _clear_cached_token(env_key: str) -> None:
|
|
"""Clear cached token for the given environment to force refresh"""
|
|
try:
|
|
cache_data = read_cache()
|
|
if cache_data and env_key in cache_data:
|
|
del cache_data[env_key]
|
|
write_cache(cache_data)
|
|
logger.info(f"Cleared cached token for environment: {env_key}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to clear cached token: {e}")
|
|
|
|
# Dependency functions
|
|
def get_credentials():
|
|
"""Get OAuth credentials from environment"""
|
|
client_id = os.getenv("CLIENT_ID", "").strip()
|
|
client_secret = os.getenv("CLIENT_SECRET", "").strip()
|
|
|
|
if not client_id or not client_secret:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Missing CLIENT_ID or CLIENT_SECRET in environment configuration"
|
|
)
|
|
|
|
return client_id, client_secret
|
|
|
|
|
|
|
|
def build_request_from_env(chilean_rut: str, request_type: str = "personal") -> EquifaxRequest:
|
|
"""Build full Equifax request from environment variables, only overriding the RUT"""
|
|
logger.debug("Building request from environment variables")
|
|
|
|
# Get all required values from environment
|
|
bill_to = os.getenv("BILL_TO", "").strip()
|
|
ship_to = os.getenv("SHIP_TO", "").strip()
|
|
product_name_empresarial = os.getenv("PRODUCT_NAME_EMPRESARIAL", "").strip()
|
|
product_orch_empresarial = os.getenv("PRODUCT_ORCH_EMPRESARIAL", "").strip()
|
|
product_name_persona = os.getenv("PRODUCT_NAME_PERSONA", "").strip()
|
|
product_orch_persona = os.getenv("PRODUCT_ORCH_PERSONA", "").strip()
|
|
|
|
configuration = os.getenv("CONFIGURATION", "Config").strip()
|
|
customer = os.getenv("CUSTOMER", "").strip()
|
|
model = os.getenv("MODEL", "").strip()
|
|
chilean_serial = os.getenv("CHILEAN_SERIAL", "").strip()
|
|
|
|
# Select product based on request type
|
|
if request_type and request_type.lower() == "empresarial":
|
|
product_name = product_name_empresarial
|
|
product_orch = product_orch_empresarial
|
|
else:
|
|
# Default to personal
|
|
product_name = product_name_persona
|
|
product_orch = product_orch_persona
|
|
|
|
# Validate required environment variables
|
|
missing = []
|
|
if not bill_to:
|
|
missing.append("BILL_TO")
|
|
if not ship_to:
|
|
missing.append("SHIP_TO")
|
|
if not product_name:
|
|
missing.append("PRODUCT_NAME")
|
|
if not product_orch:
|
|
missing.append("PRODUCT_ORCH")
|
|
if not customer:
|
|
missing.append("CUSTOMER")
|
|
if not model:
|
|
missing.append("MODEL")
|
|
|
|
# Only validate CHILEAN_SERIAL if request type is not empresarial
|
|
if request_type != "empresarial" and not chilean_serial:
|
|
missing.append("CHILEAN_SERIAL")
|
|
|
|
if missing:
|
|
error_msg = f"Missing required environment variables: {', '.join(missing)}"
|
|
logger.error(error_msg)
|
|
raise HTTPException(status_code=500, detail=error_msg)
|
|
|
|
# Build personal information
|
|
personal_info = PersonalInformation(chileanRut=chilean_rut)
|
|
|
|
# Build full request
|
|
request = EquifaxRequest(
|
|
applicants=Applicants(
|
|
primaryConsumer=PrimaryConsumer(
|
|
personalInformation=personal_info
|
|
)
|
|
),
|
|
productData=ProductData(
|
|
billTo=bill_to,
|
|
shipTo=ship_to,
|
|
productName=product_name,
|
|
productOrch=product_orch,
|
|
configuration=configuration,
|
|
customer=customer,
|
|
model=model
|
|
),
|
|
requestType=request_type
|
|
)
|
|
|
|
logger.info("Built request from environment for RUT: %s, type: %s. Request: %s", chilean_rut, request_type, request.model_dump())
|
|
return request
|
|
|
|
# API Endpoints
|
|
@app.get("/", response_model=HealthResponse)
|
|
async def root():
|
|
"""Health check endpoint"""
|
|
return HealthResponse(
|
|
status="healthy",
|
|
timestamp=datetime.now().isoformat(),
|
|
version="1.0.0"
|
|
)
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
async def health_check():
|
|
"""Detailed health check"""
|
|
return HealthResponse(
|
|
status="healthy",
|
|
timestamp=datetime.now().isoformat(),
|
|
version="1.0.0"
|
|
)
|
|
|
|
@app.post("/token/{environment}", response_model=TokenResponse)
|
|
async def get_oauth_token(environment: str):
|
|
"""Get OAuth token for specified environment"""
|
|
if environment not in ["uat", "prod"]:
|
|
raise HTTPException(status_code=400, detail="Environment must be 'uat' or 'prod'")
|
|
|
|
try:
|
|
client_id, client_secret = get_credentials()
|
|
scope = "https://api.latam.equifax.com/datos-comerciales/transaction"
|
|
|
|
# Check if we have a cached token
|
|
cached_token = get_cached_token(environment)
|
|
if cached_token:
|
|
return TokenResponse(
|
|
access_token=cached_token,
|
|
expires_in=3600, # Approximate
|
|
cached=True
|
|
)
|
|
|
|
# Fetch new token
|
|
token = fetch_token(environment, client_id, client_secret, scope)
|
|
return TokenResponse(
|
|
access_token=token,
|
|
expires_in=3600, # Approximate
|
|
cached=False
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get token: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get token: {str(e)}")
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/transaction/{environment}/rut-only",
|
|
summary="Execute Equifax Transaction (RUT Only)",
|
|
description="Execute Equifax transaction with only chileanRut required. All productData and personalData fields are automatically populated from environment variables (.env file).",
|
|
response_description="Equifax transaction result")
|
|
async def execute_rut_only_transaction(environment: str, request: RutOnlyTransactionRequest):
|
|
"""
|
|
Execute Equifax transaction with only RUT - all other data from .env
|
|
|
|
This endpoint simplifies the request by only requiring:
|
|
- chileanRut: The Chilean RUT number
|
|
- requestType: Optional, defaults to 'personal'
|
|
|
|
All other fields (productData, personalData) are automatically populated from environment variables.
|
|
"""
|
|
#print the json body of the request
|
|
logger.info(f"Received RUT-only request: {request.model_dump()}")
|
|
|
|
if environment not in ["uat", "prod"]:
|
|
raise HTTPException(status_code=400, detail="Environment must be 'uat' or 'prod'")
|
|
|
|
try:
|
|
# Build full request from environment variables
|
|
full_request = build_request_from_env(request.chileanRut, request.requestType)
|
|
|
|
client_id, client_secret = get_credentials()
|
|
scope = "https://api.latam.equifax.com/datos-comerciales/transaction"
|
|
|
|
# Get token
|
|
token = get_token(environment, client_id, client_secret, scope)
|
|
|
|
# Convert to dict for API call
|
|
payload = full_request.model_dump()
|
|
|
|
# Execute transaction
|
|
result = execute_transaction(environment, token, payload)
|
|
# logger.info("Transaction result: %s", json.dumps(result, indent=2))
|
|
return result
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"RUT-only transaction failed: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Transaction failed: {str(e)}")
|
|
|
|
@app.delete("/cache")
|
|
async def clear_token_cache():
|
|
"""Clear the token cache"""
|
|
try:
|
|
if TOKEN_CACHE_FILE.exists():
|
|
TOKEN_CACHE_FILE.unlink()
|
|
logger.info("Token cache cleared")
|
|
return {"message": "Token cache cleared successfully"}
|
|
else:
|
|
return {"message": "No token cache found"}
|
|
except Exception as e:
|
|
logger.error(f"Failed to clear cache: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to clear cache: {str(e)}")
|
|
|
|
@app.get("/cache")
|
|
async def get_cache_status():
|
|
"""Get current cache status"""
|
|
try:
|
|
cache = read_cache()
|
|
if not cache:
|
|
return {"status": "empty", "entries": 0}
|
|
|
|
status_info = {"status": "active", "entries": len(cache), "environments": {}}
|
|
|
|
current_time = now()
|
|
for env_key, token_data in cache.items():
|
|
expires_at = token_data.get("expires_at", 0)
|
|
time_until_expiry = expires_at - current_time
|
|
|
|
status_info["environments"][env_key] = {
|
|
"expires_in_seconds": max(0, time_until_expiry),
|
|
"expires_at": datetime.fromtimestamp(expires_at).isoformat() if expires_at > 0 else None,
|
|
"valid": time_until_expiry > 60 # 60s buffer
|
|
}
|
|
|
|
return status_info
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get cache status: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get cache status: {str(e)}")
|
|
|
|
@app.delete("/transaction-cache")
|
|
async def clear_transaction_cache_endpoint():
|
|
"""Clear all cached transaction results"""
|
|
try:
|
|
if cache_collection is None:
|
|
raise HTTPException(status_code=503, detail="MongoDB cache not available")
|
|
|
|
deleted_count = clear_transaction_cache()
|
|
return {
|
|
"message": "Transaction cache cleared successfully",
|
|
"deleted_count": deleted_count,
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Failed to clear transaction cache: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to clear transaction cache: {str(e)}")
|
|
|
|
@app.get("/transaction-cache/stats")
|
|
async def get_transaction_cache_stats():
|
|
"""Get transaction cache statistics"""
|
|
try:
|
|
if cache_collection is None:
|
|
raise HTTPException(status_code=503, detail="MongoDB cache not available")
|
|
|
|
stats = get_cache_stats()
|
|
return stats
|
|
except Exception as e:
|
|
logger.error(f"Failed to get transaction cache stats: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get transaction cache stats: {str(e)}")
|
|
|
|
@app.get("/config")
|
|
async def get_environment_config():
|
|
"""Get current environment configuration for Equifax requests"""
|
|
try:
|
|
config = {
|
|
"productData": {
|
|
"billTo": os.getenv("BILL_TO", ""),
|
|
"shipTo": os.getenv("SHIP_TO", ""),
|
|
"productName": os.getenv("PRODUCT_NAME", ""),
|
|
"productOrch": os.getenv("PRODUCT_ORCH", ""),
|
|
"configuration": os.getenv("CONFIGURATION", "Config"),
|
|
"customer": os.getenv("CUSTOMER", ""),
|
|
"model": os.getenv("MODEL", "")
|
|
},
|
|
"personalData": {
|
|
"chileanSerialNumber": os.getenv("CHILEAN_SERIAL", "")
|
|
},
|
|
"oauth": {
|
|
"clientId": os.getenv("CLIENT_ID", "")[:8] + "..." if os.getenv("CLIENT_ID") else "",
|
|
"scope": "https://api.latam.equifax.com/datos-comerciales/transaction"
|
|
},
|
|
"missing_variables": []
|
|
}
|
|
|
|
# Check for missing required variables
|
|
required_vars = ["BILL_TO", "SHIP_TO", "PRODUCT_NAME", "PRODUCT_ORCH", "CUSTOMER", "MODEL", "CLIENT_ID", "CLIENT_SECRET"]
|
|
for var in required_vars:
|
|
if not os.getenv(var, "").strip():
|
|
config["missing_variables"].append(var)
|
|
|
|
# Check CHILEAN_SERIAL only for personal requests
|
|
if not os.getenv("CHILEAN_SERIAL", "").strip():
|
|
config["missing_variables"].append("CHILEAN_SERIAL (required for personal requests)")
|
|
|
|
return config
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get environment config: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get config: {str(e)}")
|
|
|
|
# Error handlers
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(request, exc):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=ErrorResponse(
|
|
error=exc.detail,
|
|
detail=str(exc.detail),
|
|
timestamp=datetime.now().isoformat()
|
|
).model_dump()
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def general_exception_handler(request, exc):
|
|
logger.error(f"Unhandled exception: {exc}")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content=ErrorResponse(
|
|
error="Internal server error",
|
|
detail=str(exc),
|
|
timestamp=datetime.now().isoformat()
|
|
).model_dump()
|
|
)
|
|
|
|
async def cleanup_old_responses():
|
|
"""
|
|
Periodically cleans up old response files:
|
|
1. Deletes files older than 1 week.
|
|
2. For remaining files, keeps only the most recent one for each (prefix, rut) pair.
|
|
Runs every 10 minutes.
|
|
"""
|
|
while True:
|
|
try:
|
|
logger.info("Starting cleanup of old responses...")
|
|
if not EQUIFAX_RESPONSES_DIR.exists():
|
|
logger.info(f"Directory {EQUIFAX_RESPONSES_DIR} does not exist. Skipping cleanup.")
|
|
else:
|
|
files_by_group: Dict[str, list] = {}
|
|
one_week_ago = time.time() - (7 * 24 * 60 * 60)
|
|
deleted_count = 0
|
|
|
|
# Group files and filter old ones
|
|
for file_path in EQUIFAX_RESPONSES_DIR.glob("*.json"):
|
|
try:
|
|
filename = file_path.stem
|
|
# Expecting format: {prefix}_{rut}_{timestamp}
|
|
# Example: equifax_76042014-K_2026-01-10T15-08-11-426Z
|
|
parts = filename.rsplit('_', 1)
|
|
if len(parts) != 2:
|
|
continue
|
|
|
|
base_name, timestamp_str = parts
|
|
|
|
# Parse timestamp
|
|
# Format: YYYY-MM-DDTHH-MM-SS-mmmZ
|
|
try:
|
|
date_part, time_part = timestamp_str.split('T')
|
|
time_part = time_part.rstrip('Z')
|
|
time_parts = time_part.split('-')
|
|
|
|
hour = int(time_parts[0])
|
|
minute = int(time_parts[1])
|
|
second = int(time_parts[2])
|
|
microsecond = int(time_parts[3]) * 1000
|
|
|
|
date_parts = date_part.split('-')
|
|
year = int(date_parts[0])
|
|
month = int(date_parts[1])
|
|
day = int(date_parts[2])
|
|
|
|
dt = datetime(year, month, day, hour, minute, second, microsecond)
|
|
timestamp = dt.timestamp()
|
|
except Exception:
|
|
# If parsing fails, skip or try alternative
|
|
logger.debug(f"Failed to parse timestamp from {filename}")
|
|
continue
|
|
|
|
# Check if older than 1 week
|
|
if timestamp < one_week_ago:
|
|
try:
|
|
file_path.unlink()
|
|
deleted_count += 1
|
|
logger.info(f"Deleted old response file (older than 1 week): {file_path.name}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete {file_path}: {e}")
|
|
continue
|
|
|
|
if base_name not in files_by_group:
|
|
files_by_group[base_name] = []
|
|
|
|
files_by_group[base_name].append((timestamp, file_path))
|
|
except Exception as e:
|
|
logger.warning(f"Error processing file {file_path}: {e}")
|
|
|
|
# Process groups
|
|
for base_name, file_list in files_by_group.items():
|
|
# Sort by timestamp descending (newest first)
|
|
file_list.sort(key=lambda x: x[0], reverse=True)
|
|
|
|
# Keep the first one, delete the rest
|
|
for _, file_to_delete in file_list[1:]:
|
|
try:
|
|
file_to_delete.unlink()
|
|
deleted_count += 1
|
|
logger.info(f"Deleted old response file (duplicate): {file_to_delete.name}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete {file_to_delete}: {e}")
|
|
|
|
if deleted_count > 0:
|
|
logger.info(f"Cleanup finished. Deleted {deleted_count} old files.")
|
|
else:
|
|
logger.info("Cleanup finished. No files to delete.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during cleanup: {e}")
|
|
|
|
# Wait for 10 minutes (600 seconds)
|
|
await asyncio.sleep(600)
|
|
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
asyncio.create_task(cleanup_old_responses())
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
#print(os.environ)
|
|
uvicorn.run(app, host="localhost", port=8022)
|