Spaces:
Runtime error
Runtime error
File size: 11,084 Bytes
6391dfd a2386f0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 | """
Merchant sync handler for MongoDB to PostgreSQL synchronization.
"""
from typing import Dict, Any, Optional
from sqlalchemy.ext.asyncio import AsyncConnection
from sqlalchemy import text
from motor.motor_asyncio import AsyncIOMotorDatabase
from datetime import datetime
from dateutil import parser as date_parser
from app.core.logging import get_logger
from app.sync.common.handler import SyncHandler
from app.sync.merchants.models import MERCHANT_FIELD_MAPPING, MERCHANT_REQUIRED_FIELDS
from app.constants import SCM_MERCHANTS_COLLECTION
logger = get_logger(__name__)
class MerchantSyncHandler(SyncHandler):
"""
Handler for syncing merchant data from MongoDB to PostgreSQL.
Implements entity-specific logic for merchant synchronization including
field mapping, validation, and upsert operations.
"""
def __init__(self):
super().__init__(entity_type="merchant")
async def fetch_from_mongodb(
self,
entity_id: str,
mongo_db: Any
) -> Optional[Dict[str, Any]]:
"""
Fetch merchant from MongoDB by merchant_id.
Args:
entity_id: merchant_id to fetch
mongo_db: MongoDB database instance
Returns:
Merchant document or None if not found
"""
try:
collection = mongo_db[SCM_MERCHANTS_COLLECTION]
merchant = await collection.find_one({"merchant_id": entity_id})
return merchant
except Exception as e:
logger.error(
"Error fetching merchant from MongoDB",
exc_info=e,
extra={
"entity_type": self.entity_type,
"entity_id": entity_id
}
)
raise
def get_field_mapping(self) -> Dict[str, str]:
"""
Get field mapping from MongoDB to PostgreSQL.
Returns:
Dictionary mapping MongoDB field names to PostgreSQL column names
"""
return MERCHANT_FIELD_MAPPING
def validate_required_fields(self, entity: Dict[str, Any]) -> bool:
"""
Validate that all required fields are present in merchant document.
Required fields: merchant_id, merchant_code, merchant_type, status
Args:
entity: Merchant document from MongoDB
Returns:
True if all required fields present, False otherwise
"""
missing_fields = []
for field in MERCHANT_REQUIRED_FIELDS:
if field not in entity or entity[field] is None:
missing_fields.append(field)
if missing_fields:
logger.error(
"Merchant missing required fields",
extra={
"entity_type": self.entity_type,
"entity_id": entity.get("merchant_id"),
"missing_fields": missing_fields
}
)
return False
return True
def transform_field_value(self, field_name: str, value: Any) -> Any:
"""
Transform field value for PostgreSQL.
Handles type conversions and nested field extraction.
Args:
field_name: Name of the field
value: Value from MongoDB
Returns:
Transformed value for PostgreSQL
"""
if value is None:
return None
# Convert merchant_type enum to string if needed
if field_name == "merchant_type" and hasattr(value, 'value'):
return value.value
# Convert status enum to string if needed
if field_name == "status" and hasattr(value, 'value'):
return value.value
# Handle datetime objects - convert ISO strings to datetime
if field_name in ["created_at", "updated_at", "synced_at"]:
if isinstance(value, str):
try:
return date_parser.isoparse(value)
except Exception as e:
logger.warning(
f"Failed to parse datetime field {field_name}: {value}",
exc_info=e
)
return None
elif isinstance(value, datetime):
return value
return value
def extract_nested_fields(self, entity: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract nested fields from merchant document.
Extracts city, state, and gst_number from nested structures:
- city and state from contact.city and contact.state
- gst_number from kyc.gst_number
- Ensures created_at/updated_at have default values
Args:
entity: Merchant document from MongoDB
Returns:
Dictionary with flattened fields
"""
flattened = {}
# Extract city and state from contact
if "contact" in entity and entity["contact"]:
contact = entity["contact"]
flattened["city"] = contact.get("city")
flattened["state"] = contact.get("state")
else:
flattened["city"] = None
flattened["state"] = None
# Extract gst_number from kyc
if "kyc" in entity and entity["kyc"]:
kyc = entity["kyc"]
flattened["gst_number"] = kyc.get("gst_number")
else:
flattened["gst_number"] = None
# Ensure timestamps have default values if missing
now = datetime.utcnow()
if "created_at" not in entity or entity["created_at"] is None:
flattened["created_at"] = now
if "updated_at" not in entity or entity["updated_at"] is None:
flattened["updated_at"] = now
return flattened
async def upsert_to_postgres(
self,
entity: Dict[str, Any],
pg_conn: AsyncConnection
) -> bool:
"""
Upsert merchant to PostgreSQL trans.merchants_ref table.
Performs timestamp-based conflict resolution:
- If record doesn't exist, insert
- If MongoDB updated_at >= PostgreSQL updated_at, update
- Otherwise, skip update
Args:
entity: Merchant document from MongoDB
pg_conn: PostgreSQL connection
Returns:
True if upsert successful, False otherwise
"""
try:
merchant_id = entity["merchant_id"]
updated_at = entity.get("updated_at") or entity.get("created_at")
# Check timestamp conflict
should_update = await self.check_timestamp_conflict(
entity_id=merchant_id,
mongo_updated_at=updated_at,
pg_conn=pg_conn,
table_name="trans.merchants_ref",
id_column="merchant_id"
)
if not should_update:
logger.debug(
"Skipping merchant sync due to timestamp conflict",
extra={
"entity_type": self.entity_type,
"entity_id": merchant_id,
"mongo_updated_at": updated_at
}
)
return True # Not an error, just skipped
# Extract nested fields
nested_fields = self.extract_nested_fields(entity)
# Merge nested fields into entity for mapping
entity_with_nested = {**entity, **nested_fields}
# Map fields
mapped_entity = self.map_fields(entity_with_nested)
# Build UPSERT query
columns = list(mapped_entity.keys())
placeholders = [f":{col}" for col in columns]
# Build UPDATE clause (exclude primary key)
update_columns = [col for col in columns if col != "merchant_id"]
update_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in update_columns])
query = text(f"""
INSERT INTO trans.merchants_ref ({', '.join(columns)})
VALUES ({', '.join(placeholders)})
ON CONFLICT (merchant_id)
DO UPDATE SET {update_clause}
""")
await pg_conn.execute(query, mapped_entity)
logger.debug(
"Merchant upserted to PostgreSQL",
extra={
"entity_type": self.entity_type,
"entity_id": merchant_id
}
)
return True
except Exception as e:
logger.error(
"Error upserting merchant to PostgreSQL",
exc_info=e,
extra={
"entity_type": self.entity_type,
"entity_id": entity.get("merchant_id"),
"error": str(e)
}
)
raise
async def delete_from_postgres(
self,
entity_id: str,
pg_conn: AsyncConnection
) -> bool:
"""
Delete merchant from PostgreSQL trans.merchants_ref table.
Args:
entity_id: merchant_id to delete
pg_conn: PostgreSQL connection
Returns:
True if delete successful, False otherwise
"""
try:
merchant_id = entity_id
# Delete from PostgreSQL
query = text("DELETE FROM trans.merchants_ref WHERE merchant_id = :id")
result = await pg_conn.execute(query, {"id": merchant_id})
rows_deleted = result.rowcount
if rows_deleted > 0:
logger.info(
f"Merchant deleted from PostgreSQL",
extra={
"entity_type": self.entity_type,
"entity_id": merchant_id,
"rows_deleted": rows_deleted
}
)
return True
else:
logger.warning(
f"Merchant not found in PostgreSQL for deletion",
extra={
"entity_type": self.entity_type,
"entity_id": merchant_id
}
)
return True
except Exception as e:
logger.error(
"Error deleting merchant from PostgreSQL",
exc_info=e,
extra={
"entity_type": self.entity_type,
"entity_id": entity_id,
"error": str(e)
}
)
raise
|