"""Magento repository implementations.""" from __future__ import annotations from datetime import date, datetime, timedelta from typing import Any, Dict, Iterable, List, Optional, Tuple from magento.sync_overrides import OverrideSet from sqlalchemy import delete, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from db.models import ( MagentoAttributeCache, MagentoAttributeOptionCache, MagentoAttributeOptionRegistry, MagentoAttributeRegistry, MagentoAttributeSetAttributes, MagentoAttributeSetRegistry, MagentoBaselineSyncRun, MagentoCatalogState, MagentoCategoryCache, MagentoCategoryRegistry, MagentoConnection, MagentoMediaMap, MagentoProductOverride, MagentoProductVersion, MagentoSourceRegistry, MagentoStockRegistry, MagentoStockSourceLinks, MagentoStoreRegistry, MagentoSyncJob, MagentoSyncItem, MagentoSyncPending, MagentoSyncState, MagentoSyncNotification, MagentoSyncQueue, PlytixFeedEvent, ) from db.schemas import ( MagentoConnectionRead, MagentoConnectionUpsert, MagentoSyncItemCreate, MagentoSyncJobCreate, ) def _connection_to_read(row: MagentoConnection) -> MagentoConnectionRead: return MagentoConnectionRead( id=row.id, tenant_id=row.tenant_id, store_base_url=row.store_base_url, magento_api_base_url=row.magento_api_base_url, internal_api_base_url=row.internal_api_base_url, internal_host_header=row.internal_host_header, prefer_internal_api=bool(row.prefer_internal_api), rest_base_path=row.rest_base_path, store_code=row.store_code, environment=row.environment, status=row.status, last_verified_at=row.last_verified_at, last_error=row.last_error, created_at=row.created_at, ) class SqlAlchemyMagentoConnectionRepository: def __init__(self, session: Session) -> None: self._session = session def get_by_id(self, connection_id: int) -> Optional[MagentoConnectionRead]: row = self._session.get(MagentoConnection, connection_id) if row is None: return None return _connection_to_read(row) def get_for_sync(self, connection_id: int) -> Optional[dict]: """Return full connection dict including secrets (for sync use only).""" row = self._session.get(MagentoConnection, connection_id) if row is None: return None base = row.store_base_url.rstrip("/") public_api_base = row.magento_api_base_url or f"{base}/rest" internal_api_base = (row.internal_api_base_url or "").rstrip("/") or None prefer_internal_api = bool(row.prefer_internal_api and internal_api_base) api_base = internal_api_base if prefer_internal_api else public_api_base base_path = row.rest_base_path or "V1" # Include store_code in the path so API calls target the correct store view. # Without it, calls go to admin scope (store view 0). rest_base_path = f"{row.store_code}/{base_path}" if row.store_code else base_path return { "id": row.id, "store_base_url": row.store_base_url, "magento_api_base_url": api_base, "public_magento_api_base_url": public_api_base, "internal_api_base_url": internal_api_base, "internal_host_header": row.internal_host_header, "prefer_internal_api": prefer_internal_api, "rest_base_path": rest_base_path, "store_code": row.store_code, "consumer_key": row.consumer_key, "consumer_secret": row.consumer_secret, "access_token": row.access_token, "access_token_secret": row.access_token_secret, "signature_method": row.signature_method or "HMAC-SHA256", } def get_active_connection( self, tenant_id: Optional[str], environment: str ) -> Optional[MagentoConnectionRead]: stmt = select(MagentoConnection).where( MagentoConnection.environment == environment, MagentoConnection.status == "active", ) if tenant_id: stmt = stmt.where(MagentoConnection.tenant_id == tenant_id) row = self._session.scalars(stmt).first() if row is None: return None return _connection_to_read(row) def upsert_connection(self, payload: MagentoConnectionUpsert) -> int: tenant = payload.tenant_id or "" stmt = select(MagentoConnection).where( MagentoConnection.store_base_url == payload.store_base_url, MagentoConnection.environment == payload.environment, ) if tenant: stmt = stmt.where(MagentoConnection.tenant_id == tenant) else: stmt = stmt.where( (MagentoConnection.tenant_id == "") | (MagentoConnection.tenant_id.is_(None)) ) row = self._session.scalars(stmt).first() data = payload.model_dump(exclude_none=True) # Handle protected_fields as list for JSONB if "protected_fields" in data and data["protected_fields"] is not None: data["protected_fields"] = data["protected_fields"] # keep as list if "scopes" in data and isinstance(data["scopes"], dict): pass # keep as dict if row: for k, v in data.items(): setattr(row, k, v) row.updated_at = datetime.utcnow() self._session.add(row) self._session.flush() return int(row.id) row = MagentoConnection( tenant_id=payload.tenant_id or "", store_base_url=payload.store_base_url, magento_api_base_url=payload.magento_api_base_url or f"{payload.store_base_url.rstrip('/')}/rest", internal_api_base_url=payload.internal_api_base_url, internal_host_header=payload.internal_host_header, prefer_internal_api=bool(payload.prefer_internal_api), rest_base_path=payload.rest_base_path, store_code=payload.store_code, environment=payload.environment, consumer_key=payload.consumer_key, consumer_secret=payload.consumer_secret, access_token=payload.access_token, access_token_secret=payload.access_token_secret, scopes=payload.scopes, status=payload.status, magento_version=payload.magento_version, is_msi_enabled=payload.is_msi_enabled, default_website_id=payload.default_website_id, default_store_id=payload.default_store_id, protected_fields=payload.protected_fields, signature_method=payload.signature_method or "HMAC-SHA256", ) self._session.add(row) self._session.flush() return int(row.id) def list_connections(self, tenant_id: Optional[str] = None) -> List[MagentoConnectionRead]: stmt = select(MagentoConnection).order_by(MagentoConnection.id.desc()) if tenant_id: stmt = stmt.where(MagentoConnection.tenant_id == tenant_id) rows = self._session.scalars(stmt).all() return [_connection_to_read(r) for r in rows] def update_verification( self, connection_id: int, *, success: bool, error: Optional[str] = None ) -> None: row = self._session.get(MagentoConnection, connection_id) if row is None: return row.last_verified_at = datetime.utcnow() if success else row.last_verified_at row.last_error = None if success else error row.status = "active" if success else "expired" self._session.add(row) class SqlAlchemyMagentoSyncRepository: def __init__(self, session: Session) -> None: self._session = session def create_sync_job(self, payload: MagentoSyncJobCreate) -> int: row = MagentoSyncJob( connection_id=payload.connection_id, status="started", dry_run=payload.dry_run, total_count=payload.total_count, ) self._session.add(row) self._session.flush() return int(row.id) def update_sync_job(self, job_id: int, **kwargs: object) -> None: row = self._session.get(MagentoSyncJob, job_id) if row is None: return for k, v in kwargs.items(): if hasattr(row, k): setattr(row, k, v) self._session.add(row) def get_sync_job(self, job_id: int) -> Optional[Dict[str, Any]]: row = self._session.get(MagentoSyncJob, job_id) if row is None: return None return { "id": row.id, "connection_id": row.connection_id, "status": row.status, "started_at": row.started_at, "finished_at": row.finished_at, "success_count": row.success_count, "error_count": row.error_count, } def add_sync_items(self, items: Iterable[MagentoSyncItemCreate]) -> None: for item in items: data = item.model_dump() row = MagentoSyncItem(**data) self._session.add(row) self._session.flush() def get_job_items(self, job_id: int) -> List[dict]: stmt = select(MagentoSyncItem).where(MagentoSyncItem.job_id == job_id) rows = self._session.scalars(stmt).all() return [ { "id": r.id, "sku": r.sku, "action": r.action, "status": r.status, "error": r.error, "http_status": r.http_status, } for r in rows ] def get_failed_sync_items( self, connection_id: int, *, action: Optional[str] = None, error_like: Optional[str] = None, limit: int = 500, unique_skus: bool = True, ) -> List[Dict[str, Any]]: """Return failed sync items for a connection. unique_skus=True: one row per SKU (most recent failure per SKU). """ stmt = ( select(MagentoSyncItem, MagentoSyncJob.started_at) .join(MagentoSyncJob, MagentoSyncJob.id == MagentoSyncItem.job_id) .where( MagentoSyncJob.connection_id == connection_id, MagentoSyncItem.status == "failed", MagentoSyncItem.error.isnot(None), ) ) if action: stmt = stmt.where(MagentoSyncItem.action == action) if error_like: stmt = stmt.where(MagentoSyncItem.error.ilike(f"%{error_like}%")) stmt = stmt.order_by(MagentoSyncJob.started_at.desc(), MagentoSyncItem.id.desc()).limit( limit * 3 if unique_skus else limit ) rows = self._session.execute(stmt).all() result: List[Dict[str, Any]] = [] seen_skus: set = set() for r in rows: sku = r[0].sku or "" if unique_skus and sku in seen_skus: continue if unique_skus: seen_skus.add(sku) result.append({ "id": r[0].id, "job_id": r[0].job_id, "sku": r[0].sku, "action": r[0].action, "error": r[0].error, "http_status": r[0].http_status, "started_at": r[1].isoformat() if r[1] else None, }) if len(result) >= limit: break return result def get_pending_items(self, job_id: int) -> List[dict]: """Items with status=pending, ordered by id for resumable execution.""" stmt = ( select(MagentoSyncItem) .where( MagentoSyncItem.job_id == job_id, MagentoSyncItem.status == "pending", ) .order_by(MagentoSyncItem.id) ) rows = self._session.scalars(stmt).all() return [ { "id": r.id, "sku": r.sku, "action": r.action, "payload": r.payload, "idempotency_key": r.idempotency_key, } for r in rows ] def update_sync_item( self, item_id: int, *, status: Optional[str] = None, error: Optional[str] = None, duration_ms: Optional[int] = None, http_status: Optional[int] = None, payload: Optional[dict] = None, remote_image_urls: Optional[list] = None, version_id: Optional[int] = None, magento_response: Optional[dict] = None, ) -> None: """Update a sync item after execution for resumable sync.""" row = self._session.get(MagentoSyncItem, item_id) if row is None: return if status is not None: row.status = status if error is not None: row.error = error if duration_ms is not None: row.duration_ms = duration_ms if http_status is not None: row.http_status = http_status if payload is not None: row.payload = payload if remote_image_urls is not None: row.remote_image_urls = remote_image_urls if version_id is not None: row.version_id = version_id if magento_response is not None: row.magento_response = magento_response self._session.add(row) def mark_failed_items_resolved( self, connection_id: int, skus: Optional[List[str]] = None, *, state_repo: Optional["SqlAlchemyMagentoSyncStateRepository"] = None, ) -> int: """ Mark failed sync items as resolved (succeeded) for given SKUs. - Updates magento_sync_items: status=succeeded, error=None - Clears last_error in magento_sync_state for those SKUs Use when products are confirmed active on Magento despite prior sync failures. Returns the number of magento_sync_items updated. """ job_ids_stmt = select(MagentoSyncJob.id).where( MagentoSyncJob.connection_id == connection_id ) # Capture affected SKUs before update skus_stmt = ( select(MagentoSyncItem.sku) .where( MagentoSyncItem.job_id.in_(job_ids_stmt), MagentoSyncItem.status == "failed", ) .distinct() ) if skus: skus_stmt = skus_stmt.where(MagentoSyncItem.sku.in_(skus)) skus_to_clear = [r[0] for r in self._session.execute(skus_stmt).all() if r[0]] stmt = ( update(MagentoSyncItem) .where( MagentoSyncItem.job_id.in_(job_ids_stmt), MagentoSyncItem.status == "failed", ) .values(status="succeeded", error=None) ) if skus: stmt = stmt.where(MagentoSyncItem.sku.in_(skus)) result = self._session.execute(stmt) items_updated = result.rowcount # Clear last_error in magento_sync_state for affected SKUs if state_repo and skus_to_clear: for sku in skus_to_clear: state_repo.upsert_state(connection_id, sku, last_error="") return items_updated def get_datewise_report( self, connection_id: int, date_from: date, date_to: date ) -> List[Dict[str, Any]]: """Jobs and items grouped by day for reporting.""" from sqlalchemy import func stmt = ( select( func.date(MagentoSyncJob.started_at).label("day"), func.count(MagentoSyncJob.id).label("jobs_count"), func.sum(MagentoSyncJob.success_count).label("total_success"), func.sum(MagentoSyncJob.error_count).label("total_failed"), ) .where( MagentoSyncJob.connection_id == connection_id, MagentoSyncJob.started_at.isnot(None), func.date(MagentoSyncJob.started_at) >= date_from, func.date(MagentoSyncJob.started_at) <= date_to, ) .group_by(func.date(MagentoSyncJob.started_at)) .order_by(func.date(MagentoSyncJob.started_at)) ) rows = self._session.execute(stmt).all() result: List[Dict[str, Any]] = [] for r in rows: day_val = getattr(r, "day", None) or (r[0] if len(r) > 0 else None) day_str = day_val.isoformat() if day_val else "" jobs_cnt = getattr(r, "jobs_count", None) or (r[1] if len(r) > 1 else None) or 0 total_ok = getattr(r, "total_success", None) or (r[2] if len(r) > 2 else None) or 0 total_fail = getattr(r, "total_failed", None) or (r[3] if len(r) > 3 else None) or 0 result.append({ "day": day_str, "jobs_count": jobs_cnt, "total_success": int(total_ok), "total_failed": int(total_fail), "breakdown": {}, "failed_skus": [], }) for rec in result: day_str = rec["day"] try: day_date = date.fromisoformat(day_str) except ValueError: continue job_ids_stmt = ( select(MagentoSyncJob.id).where( MagentoSyncJob.connection_id == connection_id, func.date(MagentoSyncJob.started_at) == day_date, ) ) job_ids = [rid for (rid,) in self._session.execute(job_ids_stmt).all()] if not job_ids: continue item_stmt = select( MagentoSyncItem.action, MagentoSyncItem.status, MagentoSyncItem.sku, MagentoSyncItem.error ).where(MagentoSyncItem.job_id.in_(job_ids)) items = self._session.execute(item_stmt).all() breakdown: Dict[str, Dict[str, int]] = {} failed_skus: List[Dict[str, str]] = [] for row in items: action, status, sku, err = row if len(row) >= 4 else (row[0], row[1], row[2], row[3] if len(row) > 3 else None) action = action or "unknown" breakdown.setdefault(action, {"success": 0, "failed": 0}) if status == "succeeded": breakdown[action]["success"] += 1 else: breakdown[action]["failed"] += 1 failed_skus.append({"sku": sku or "", "error": (err or "")[:200]}) rec["breakdown"] = breakdown rec["failed_skus"] = failed_skus[:50] return result class SqlAlchemyMagentoSyncPendingRepository: """Connection-scoped pending work. UPSERT on plan; update existing pending with new payload.""" def __init__(self, session: Session) -> None: self._session = session def upsert_pending( self, connection_id: int, sku: str, action: str, *, payload: Optional[dict] = None, pass_number: int = 1, idempotency_key: Optional[str] = None, ) -> None: """Insert or update pending. If row exists (any status), update payload and set status=pending.""" from sqlalchemy.dialects.postgresql import insert now = datetime.utcnow() stmt = insert(MagentoSyncPending).values( connection_id=connection_id, sku=sku, action=action, payload=payload, status="pending", pass_number=pass_number, idempotency_key=idempotency_key, updated_at=now, ) stmt = stmt.on_conflict_do_update( index_elements=["connection_id", "sku", "action"], set_={ "payload": payload, "status": "pending", "pass_number": pass_number, "idempotency_key": idempotency_key, "updated_at": now, "error": None, }, ) self._session.execute(stmt) def get_pending_items(self, connection_id: int, *, skus: Optional[Iterable[str]] = None) -> List[dict]: """Items with status=pending, ordered by pass_number, sku, action.""" stmt = ( select(MagentoSyncPending) .where( MagentoSyncPending.connection_id == connection_id, MagentoSyncPending.status == "pending", ) .order_by(MagentoSyncPending.pass_number, MagentoSyncPending.sku, MagentoSyncPending.action) ) if skus is not None: wanted = sorted({str(s).strip() for s in skus if str(s).strip()}) if wanted: stmt = stmt.where(MagentoSyncPending.sku.in_(wanted)) rows = self._session.execute(stmt).scalars().all() return [ { "id": r.id, "sku": r.sku, "action": r.action, "payload": r.payload, "idempotency_key": r.idempotency_key, } for r in rows ] def update_pending_item( self, item_id: int, *, status: Optional[str] = None, error: Optional[str] = None, duration_ms: Optional[int] = None, http_status: Optional[int] = None, ) -> None: """Update pending item after execution.""" row = self._session.get(MagentoSyncPending, item_id) if row is None: return if status is not None: row.status = status if error is not None: row.error = error if duration_ms is not None: row.duration_ms = duration_ms if http_status is not None: row.http_status = http_status row.updated_at = datetime.utcnow() self._session.add(row) def count_pending(self, connection_id: int) -> int: """Count pending items for connection.""" from sqlalchemy import func stmt = select(func.count(MagentoSyncPending.id)).where( MagentoSyncPending.connection_id == connection_id, MagentoSyncPending.status == "pending", ) return self._session.execute(stmt).scalar() or 0 def prune_pending_not_in_plan( self, connection_id: int, planned_keys: Iterable[tuple[str, str]], ) -> int: """Drop stale pending rows (e.g. removed SKUs) not in the current plan.""" allowed = { (str(sku).strip(), str(action).strip()) for sku, action in planned_keys if str(sku).strip() and str(action).strip() } if not allowed: return 0 rows = self._session.scalars( select(MagentoSyncPending).where( MagentoSyncPending.connection_id == connection_id, MagentoSyncPending.status == "pending", ) ).all() removed = 0 for row in rows: key = (str(row.sku or "").strip(), str(row.action or "").strip()) if key not in allowed: self._session.delete(row) removed += 1 return removed class SqlAlchemyMagentoSyncStateRepository: """CRUD for magento_sync_state - incremental sync hashes per connection+sku.""" def __init__(self, session: Session) -> None: self._session = session def get_state(self, connection_id: int, sku: str) -> Optional[dict]: """Get state row for connection+sku. Returns None if not found.""" stmt = select(MagentoSyncState).where( MagentoSyncState.connection_id == connection_id, MagentoSyncState.sku == sku, ) row = self._session.scalars(stmt).first() if row is None: return None return { "id": row.id, "connection_id": row.connection_id, "sku": row.sku, "data_hash": row.data_hash, "images_hash": row.images_hash, "relations_hash": row.relations_hash, "associations_hash": row.associations_hash, "last_seen_in_feed_at": row.last_seen_in_feed_at, "last_pushed_at": row.last_pushed_at, "last_error": row.last_error, "created_at": row.created_at, "updated_at": row.updated_at, } def get_states_for_skus( self, connection_id: int, skus: List[str] ) -> dict[str, dict]: """Batch get state for multiple SKUs. Returns {sku: state_dict}.""" if not skus: return {} stmt = select(MagentoSyncState).where( MagentoSyncState.connection_id == connection_id, MagentoSyncState.sku.in_(skus), ) rows = self._session.scalars(stmt).all() return { row.sku: { "id": row.id, "connection_id": row.connection_id, "sku": row.sku, "data_hash": row.data_hash, "images_hash": row.images_hash, "relations_hash": row.relations_hash, "associations_hash": row.associations_hash, "last_seen_in_feed_at": row.last_seen_in_feed_at, "last_pushed_at": row.last_pushed_at, "last_error": row.last_error, "created_at": row.created_at, "updated_at": row.updated_at, } for row in rows } def get_all_states_for_connection(self, connection_id: int) -> Dict[str, dict]: """Return {sku: state_dict} for all SKUs tracked under a connection.""" stmt = select(MagentoSyncState).where( MagentoSyncState.connection_id == connection_id, ) rows = self._session.scalars(stmt).all() return { row.sku: { "sku": row.sku, "data_hash": row.data_hash, "images_hash": row.images_hash, "relations_hash": row.relations_hash, "associations_hash": row.associations_hash, "last_pushed_at": row.last_pushed_at.isoformat() if row.last_pushed_at else None, "last_error": row.last_error, } for row in rows } def reset_hashes( self, connection_id: int, skus: Optional[List[str]] = None, ) -> int: """Null all three hashes (and last_error) so every pass re-runs on next sync. If skus is provided, only those SKUs are reset; otherwise all SKUs for the connection. Returns the number of rows updated.""" stmt = ( update(MagentoSyncState) .where(MagentoSyncState.connection_id == connection_id) .values(data_hash=None, images_hash=None, relations_hash=None, associations_hash=None, last_error=None) ) if skus: stmt = stmt.where(MagentoSyncState.sku.in_(skus)) result = self._session.execute(stmt) return result.rowcount def upsert_state( self, connection_id: int, sku: str, *, data_hash: Optional[str] = None, images_hash: Optional[str] = None, relations_hash: Optional[str] = None, associations_hash: Optional[str] = None, last_seen_in_feed_at: Optional[datetime] = None, last_pushed_at: Optional[datetime] = None, last_error: Optional[str] = None, ) -> None: """ Insert or update state. Only updates provided fields (partial update). For hashes: pass new value to update; None means leave unchanged. last_error: pass "" to clear. """ stmt = select(MagentoSyncState).where( MagentoSyncState.connection_id == connection_id, MagentoSyncState.sku == sku, ) row = self._session.scalars(stmt).first() now = datetime.utcnow() if row is None: row = MagentoSyncState( connection_id=connection_id, sku=sku, data_hash=data_hash, images_hash=images_hash, relations_hash=relations_hash, associations_hash=associations_hash, last_seen_in_feed_at=last_seen_in_feed_at, last_pushed_at=last_pushed_at, last_error=last_error if last_error is not None else None, ) self._session.add(row) else: if data_hash is not None: row.data_hash = data_hash if images_hash is not None: row.images_hash = images_hash if relations_hash is not None: row.relations_hash = relations_hash if associations_hash is not None: row.associations_hash = associations_hash if last_seen_in_feed_at is not None: row.last_seen_in_feed_at = last_seen_in_feed_at if last_pushed_at is not None: row.last_pushed_at = last_pushed_at if last_error is not None: row.last_error = last_error if last_error else None row.updated_at = now self._session.add(row) self._session.flush() class SqlAlchemyMagentoMediaMapRepository: """CRUD for magento_media_map - image content hash → Magento entry_id.""" def __init__(self, session: Session) -> None: self._session = session def get_by_sku_url(self, connection_id: int, sku: str, remote_url: str) -> Optional[dict]: """Get mapping for sku+url. Returns None if not found.""" stmt = select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, MagentoMediaMap.remote_url == remote_url, ) row = self._session.scalars(stmt).first() if row is None: return None return { "content_sha256": row.content_sha256, "magento_entry_id": row.magento_entry_id, } def get_media_maps_for_sku(self, connection_id: int, sku: str) -> Dict[str, dict]: """Return {remote_url: row_dict} for all mappings of this SKU.""" stmt = select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, ) rows = self._session.execute(stmt).scalars().all() return { r.remote_url: { "content_sha256": r.content_sha256, "magento_entry_id": r.magento_entry_id, "magento_file": r.magento_file, } for r in rows } def get_by_sha256(self, connection_id: int, sku: str, content_sha256: str) -> Optional[dict]: """Check if we already have this content hash for this sku (reuse entry_id).""" stmt = select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, MagentoMediaMap.content_sha256 == content_sha256, ) row = self._session.scalars(stmt).first() if row is None: return None return { "remote_url": row.remote_url, "content_sha256": row.content_sha256, "magento_entry_id": row.magento_entry_id, } def upsert( self, connection_id: int, sku: str, remote_url: str, content_sha256: Optional[str] = None, magento_entry_id: Optional[int] = None, magento_file: Optional[str] = None, ) -> None: """Insert or update media map. Unique on (conn, sku, remote_url) and (conn, sku, content_sha256). Same image at different URLs must merge onto one row. Concurrent workers may race on INSERT — catch UniqueViolation inside a savepoint and re-merge without aborting the outer sync transaction. """ from sqlalchemy.exc import IntegrityError sku = str(sku or "").strip() remote_url = str(remote_url or "").strip() if not sku or not remote_url: return sha = str(content_sha256).strip() if content_sha256 else None if sha == "": sha = None last_exc: Optional[BaseException] = None for _attempt in range(3): try: with self._session.begin_nested(): self._upsert_media_map_once( connection_id, sku, remote_url, content_sha256=sha, magento_entry_id=magento_entry_id, magento_file=magento_file, ) return except IntegrityError as exc: last_exc = exc self._session.expire_all() if last_exc is not None: raise last_exc def _find_media_map_row( self, connection_id: int, sku: str, *, remote_url: Optional[str] = None, content_sha256: Optional[str] = None, ) -> Optional[MagentoMediaMap]: if remote_url: row = self._session.scalars( select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, MagentoMediaMap.remote_url == remote_url, ) ).first() if row is not None: return row if content_sha256: return self._session.scalars( select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, MagentoMediaMap.content_sha256 == content_sha256, ) ).first() return None def _apply_media_map_fields( self, row: MagentoMediaMap, *, remote_url: str, content_sha256: Optional[str], magento_entry_id: Optional[int], magento_file: Optional[str], now: datetime, ) -> None: row.remote_url = remote_url if content_sha256 is not None: row.content_sha256 = content_sha256 if magento_entry_id is not None: row.magento_entry_id = magento_entry_id if magento_file is not None: row.magento_file = magento_file row.updated_at = now self._session.add(row) def _upsert_media_map_once( self, connection_id: int, sku: str, remote_url: str, *, content_sha256: Optional[str], magento_entry_id: Optional[int], magento_file: Optional[str], ) -> None: now = datetime.utcnow() by_url = self._find_media_map_row(connection_id, sku, remote_url=remote_url) by_sha = ( self._find_media_map_row(connection_id, sku, content_sha256=content_sha256) if content_sha256 else None ) # Same content already mapped under a different URL → keep sha row, drop URL dupe. if by_url is not None and by_sha is not None and by_url.id != by_sha.id: keep = by_sha drop = by_url if magento_entry_id is None and drop.magento_entry_id and not keep.magento_entry_id: magento_entry_id = drop.magento_entry_id if magento_file is None and drop.magento_file and not keep.magento_file: magento_file = drop.magento_file self._session.delete(drop) self._session.flush() self._apply_media_map_fields( keep, remote_url=remote_url, content_sha256=content_sha256, magento_entry_id=magento_entry_id, magento_file=magento_file, now=now, ) self._session.flush() return row = by_url or by_sha if row is not None: # Updating URL-row sha onto an existing sha would UniqueViolation — merge. if ( content_sha256 and by_url is not None and row.content_sha256 and row.content_sha256 != content_sha256 ): other = self._find_media_map_row( connection_id, sku, content_sha256=content_sha256 ) if other is not None and other.id != row.id: if magento_entry_id is None and row.magento_entry_id and not other.magento_entry_id: magento_entry_id = row.magento_entry_id if magento_file is None and row.magento_file and not other.magento_file: magento_file = row.magento_file self._session.delete(row) self._session.flush() self._apply_media_map_fields( other, remote_url=remote_url, content_sha256=content_sha256, magento_entry_id=magento_entry_id, magento_file=magento_file, now=now, ) self._session.flush() return self._apply_media_map_fields( row, remote_url=remote_url, content_sha256=content_sha256, magento_entry_id=magento_entry_id, magento_file=magento_file, now=now, ) self._session.flush() return bind = self._session.get_bind() if bind is not None and bind.dialect.name == "postgresql": values = { "connection_id": connection_id, "sku": sku, "remote_url": remote_url, "content_sha256": content_sha256, "magento_entry_id": magento_entry_id, "magento_file": magento_file, "created_at": now, "updated_at": now, } stmt = pg_insert(MagentoMediaMap).values(values) set_cols = { "updated_at": now, "magento_entry_id": stmt.excluded.magento_entry_id, "magento_file": stmt.excluded.magento_file, } if content_sha256 is not None: set_cols["content_sha256"] = stmt.excluded.content_sha256 stmt = stmt.on_conflict_do_update( constraint="uq_magento_media_map_conn_sku_url", set_=set_cols, ) self._session.execute(stmt) self._session.flush() # URL upsert may leave a duplicate sha row from an older URL — consolidate. if content_sha256: by_url2 = self._find_media_map_row(connection_id, sku, remote_url=remote_url) by_sha2 = self._find_media_map_row( connection_id, sku, content_sha256=content_sha256 ) if by_url2 is not None and by_sha2 is not None and by_url2.id != by_sha2.id: if not by_url2.magento_entry_id and by_sha2.magento_entry_id: by_url2.magento_entry_id = by_sha2.magento_entry_id if not by_url2.magento_file and by_sha2.magento_file: by_url2.magento_file = by_sha2.magento_file by_url2.content_sha256 = content_sha256 by_url2.updated_at = now self._session.delete(by_sha2) self._session.flush() return self._session.add( MagentoMediaMap( connection_id=connection_id, sku=sku, remote_url=remote_url, content_sha256=content_sha256, magento_entry_id=magento_entry_id, magento_file=magento_file, ) ) self._session.flush() def list_rows_missing_magento_file( self, connection_id: int ) -> List[dict]: """Return rows where magento_entry_id IS NOT NULL and magento_file IS NULL.""" stmt = select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.magento_entry_id.isnot(None), MagentoMediaMap.magento_file.is_(None), ) rows = self._session.execute(stmt).scalars().all() return [ { "id": r.id, "connection_id": r.connection_id, "sku": r.sku, "remote_url": r.remote_url, "magento_entry_id": r.magento_entry_id, } for r in rows ] def update_magento_file_by_id(self, row_id: int, magento_file: str) -> None: """Update magento_file for a row by id.""" row = self._session.get(MagentoMediaMap, row_id) if row: row.magento_file = magento_file row.updated_at = datetime.utcnow() self._session.add(row) self._session.flush() def list_rows_for_sku(self, connection_id: int, sku: str) -> List[dict]: """Return media-map rows for one SKU with identifiers needed for cleanup/debug.""" stmt = select(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, ) rows = self._session.execute(stmt).scalars().all() return [ { "id": r.id, "sku": r.sku, "remote_url": r.remote_url, "content_sha256": r.content_sha256, "magento_entry_id": r.magento_entry_id, "magento_file": r.magento_file, "created_at": r.created_at, "updated_at": r.updated_at, } for r in rows ] def delete_for_sku_urls(self, connection_id: int, sku: str, remote_urls: List[str]) -> int: """Delete stale media-map rows for one SKU by remote_url.""" wanted = [str(url).strip() for url in remote_urls if str(url).strip()] if not wanted: return 0 stmt = delete(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, MagentoMediaMap.remote_url.in_(wanted), ) result = self._session.execute(stmt) self._session.flush() return result.rowcount or 0 def delete_all_for_sku(self, connection_id: int, sku: str) -> int: """Delete every media-map row for one SKU (used by wipe-all gallery replace).""" stmt = delete(MagentoMediaMap).where( MagentoMediaMap.connection_id == connection_id, MagentoMediaMap.sku == sku, ) result = self._session.execute(stmt) self._session.flush() return result.rowcount or 0 from channel.attribute_value_resolver import norm_option_label as _norm_option_label class SqlAlchemyMagentoAttributeCacheRepository: """Cache for Magento attributes + options. TTL-controlled. Used for SET_RELATIONS value_index.""" def __init__(self, session: Session) -> None: self._session = session def get_attributes( self, connection_id: int, codes: List[str], ttl_cutoff: datetime ) -> Dict[str, dict]: """ Get cached attributes. Returns {code: row} only for codes that exist and are fresh. codes: list of attribute_code (lowercase). ttl_cutoff: fetched_at >= ttl_cutoff means valid. """ if not codes: return {} codes_lower = [c.strip().lower() for c in codes if c.strip()] if not codes_lower: return {} stmt = select(MagentoAttributeCache).where( MagentoAttributeCache.connection_id == connection_id, MagentoAttributeCache.attribute_code.in_(codes_lower), MagentoAttributeCache.fetched_at >= ttl_cutoff, ) rows = self._session.execute(stmt).scalars().all() return { str(r.attribute_code).lower(): { "attribute_code": r.attribute_code, "attribute_id": r.attribute_id, "frontend_label": r.frontend_label, "frontend_input": r.frontend_input, "backend_type": r.backend_type, "is_user_defined": r.is_user_defined, "fetched_at": r.fetched_at, } for r in rows } def upsert_attribute( self, connection_id: int, attribute_code: str, attribute_id: int, *, frontend_label: Optional[str] = None, frontend_input: Optional[str] = None, backend_type: Optional[str] = None, is_user_defined: Optional[bool] = None, fetched_at: datetime, ) -> None: """Insert or update attribute cache row.""" code = attribute_code.strip().lower() stmt = select(MagentoAttributeCache).where( MagentoAttributeCache.connection_id == connection_id, MagentoAttributeCache.attribute_code == code, ) row = self._session.scalars(stmt).first() if row is None: row = MagentoAttributeCache( connection_id=connection_id, attribute_code=code, attribute_id=attribute_id, frontend_label=frontend_label, frontend_input=frontend_input, backend_type=backend_type, is_user_defined=is_user_defined, fetched_at=fetched_at, ) self._session.add(row) else: row.attribute_id = attribute_id row.frontend_label = frontend_label or row.frontend_label row.frontend_input = frontend_input or row.frontend_input row.backend_type = backend_type or row.backend_type row.is_user_defined = is_user_defined if is_user_defined is not None else row.is_user_defined row.fetched_at = fetched_at self._session.add(row) self._session.flush() def get_options( self, connection_id: int, attribute_code: str, ttl_cutoff: datetime ) -> Dict[str, int]: """ Get cached options: {label_norm: value_index}. Only returns if cache is fresh (fetched_at >= ttl_cutoff). """ code = attribute_code.strip().lower() stmt = select(MagentoAttributeOptionCache).where( MagentoAttributeOptionCache.connection_id == connection_id, MagentoAttributeOptionCache.attribute_code == code, MagentoAttributeOptionCache.fetched_at >= ttl_cutoff, ) rows = self._session.execute(stmt).scalars().all() return {r.option_label_norm: r.value_index for r in rows} def bulk_upsert_options( self, connection_id: int, attribute_code: str, options: List[Dict[str, Any]], fetched_at: datetime, ) -> None: """options: [{"label": "X", "value_index": 123}, ...]. value_index can be int or str.""" code = attribute_code.strip().lower() for opt in options: label = str(opt.get("label", "")).strip() vi = opt.get("value_index") if opt.get("value_index") is not None else opt.get("value") if label and vi is not None: try: value_index = int(vi) except (ValueError, TypeError): continue norm = _norm_option_label(label) if not norm: continue stmt = select(MagentoAttributeOptionCache).where( MagentoAttributeOptionCache.connection_id == connection_id, MagentoAttributeOptionCache.attribute_code == code, MagentoAttributeOptionCache.option_label_norm == norm, ) row = self._session.scalars(stmt).first() if row is None: row = MagentoAttributeOptionCache( connection_id=connection_id, attribute_code=code, option_label=label, option_label_norm=norm, value_index=value_index, fetched_at=fetched_at, ) self._session.add(row) else: row.option_label = label row.value_index = value_index row.fetched_at = fetched_at self._session.add(row) self._session.flush() def clear_for_connection(self, connection_id: int) -> Tuple[int, int]: """ Delete all attribute and option cache rows for a connection. Returns (attributes_deleted, options_deleted). Use when cache may be stale (e.g. after attribute changes in Magento). """ opts_deleted = self._session.execute( delete(MagentoAttributeOptionCache).where( MagentoAttributeOptionCache.connection_id == connection_id ) ).rowcount attrs_deleted = self._session.execute( delete(MagentoAttributeCache).where( MagentoAttributeCache.connection_id == connection_id ) ).rowcount self._session.flush() return attrs_deleted, opts_deleted class SqlAlchemyMagentoCategoryCacheRepository: """Cache for Magento category tree. Resolves category path strings to IDs.""" def __init__(self, session: Session) -> None: self._session = session def get_all(self, connection_id: int, ttl_cutoff: datetime) -> List[dict]: """Return all cached categories that are fresh.""" stmt = select(MagentoCategoryCache).where( MagentoCategoryCache.connection_id == connection_id, MagentoCategoryCache.fetched_at >= ttl_cutoff, ) rows = self._session.execute(stmt).scalars().all() return [ { "magento_category_id": r.magento_category_id, "parent_id": r.parent_id, "name": r.name, "path": r.path, "path_names": r.path_names, "level": r.level, "is_active": r.is_active, } for r in rows ] def resolve_path(self, connection_id: int, path_names: str, ttl_cutoff: datetime) -> Optional[int]: """Find category_id by exact path_names match (e.g. 'Default Category/Apparel/Tops').""" norm = path_names.strip() if not norm: return None stmt = select(MagentoCategoryCache).where( MagentoCategoryCache.connection_id == connection_id, MagentoCategoryCache.path_names == norm, MagentoCategoryCache.fetched_at >= ttl_cutoff, ) row = self._session.execute(stmt).scalars().first() return row.magento_category_id if row else None def resolve_name(self, connection_id: int, name: str, ttl_cutoff: datetime) -> Optional[int]: """Find category_id by leaf name (case-insensitive). Returns first active match.""" norm = name.strip().lower() if not norm: return None stmt = select(MagentoCategoryCache).where( MagentoCategoryCache.connection_id == connection_id, MagentoCategoryCache.is_active.is_(True), MagentoCategoryCache.fetched_at >= ttl_cutoff, ) rows = self._session.execute(stmt).scalars().all() for r in rows: if r.name.strip().lower() == norm: return r.magento_category_id return None def bulk_replace(self, connection_id: int, categories: List[dict], fetched_at: datetime) -> int: """Replace all cached categories for a connection. Returns count inserted.""" stmt = select(MagentoCategoryCache).where( MagentoCategoryCache.connection_id == connection_id, ) existing = self._session.execute(stmt).scalars().all() existing_map = {r.magento_category_id: r for r in existing} count = 0 for cat in categories: cat_id = cat["magento_category_id"] row = existing_map.get(cat_id) if row: row.parent_id = cat.get("parent_id") row.name = cat["name"] row.path = cat["path"] row.path_names = cat.get("path_names") row.level = cat.get("level", 0) row.is_active = cat.get("is_active", True) row.fetched_at = fetched_at else: row = MagentoCategoryCache( connection_id=connection_id, magento_category_id=cat_id, parent_id=cat.get("parent_id"), name=cat["name"], path=cat["path"], path_names=cat.get("path_names"), level=cat.get("level", 0), is_active=cat.get("is_active", True), fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def clear_for_connection(self, connection_id: int) -> int: """ Delete all category cache rows for a connection. Returns count deleted. Use when category tree changed in Magento. """ count = self._session.execute( delete(MagentoCategoryCache).where( MagentoCategoryCache.connection_id == connection_id ) ).rowcount self._session.flush() return count # --------------------------------------------------------------------------- # Baseline Registry Repositories # --------------------------------------------------------------------------- class SqlAlchemyMagentoStoreRegistryRepository: """CRUD for magento_store_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace( self, connection_id: int, rows: List[Dict[str, Any]], fetched_at: datetime ) -> int: existing_stmt = select(MagentoStoreRegistry).where( MagentoStoreRegistry.connection_id == connection_id, ) existing = { (r.website_id, r.store_id): r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for entry in rows: key = (entry["website_id"], entry.get("store_id")) row = existing.get(key) if row: row.website_code = entry.get("website_code") row.website_name = entry.get("website_name") row.store_group_id = entry.get("store_group_id") row.store_group_name = entry.get("store_group_name") row.store_code = entry.get("store_code") row.store_name = entry.get("store_name") row.is_default = entry.get("is_default", False) row.raw_json = entry.get("raw_json") row.fetched_at = fetched_at else: row = MagentoStoreRegistry( connection_id=connection_id, website_id=entry["website_id"], website_code=entry.get("website_code"), website_name=entry.get("website_name"), store_group_id=entry.get("store_group_id"), store_group_name=entry.get("store_group_name"), store_id=entry.get("store_id"), store_code=entry.get("store_code"), store_name=entry.get("store_name"), is_default=entry.get("is_default", False), raw_json=entry.get("raw_json"), fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_default_website_id(self, connection_id: int) -> Optional[int]: stmt = select(MagentoStoreRegistry).where( MagentoStoreRegistry.connection_id == connection_id, MagentoStoreRegistry.is_default.is_(True), ) row = self._session.execute(stmt).scalars().first() if row: return row.website_id stmt_any = select(MagentoStoreRegistry).where( MagentoStoreRegistry.connection_id == connection_id, ).limit(1) row = self._session.execute(stmt_any).scalars().first() return row.website_id if row else None def get_store_ids_for_default_website( self, connection_id: int, website_id: Optional[int] = None ) -> List[int]: """ Return store_ids for the default (or given) website. Used to inject baseline store IDs when the feed has 0 or invalid values. """ web_id = website_id or self.get_default_website_id(connection_id) if not web_id: return [] stmt = select(MagentoStoreRegistry).where( MagentoStoreRegistry.connection_id == connection_id, MagentoStoreRegistry.website_id == web_id, MagentoStoreRegistry.store_id.isnot(None), ) return [ int(r.store_id) for r in self._session.execute(stmt).scalars().all() if r.store_id is not None ] def get_all(self, connection_id: int) -> List[Dict[str, Any]]: stmt = select(MagentoStoreRegistry).where( MagentoStoreRegistry.connection_id == connection_id, ) rows = self._session.execute(stmt).scalars().all() return [ { "website_id": r.website_id, "website_code": r.website_code, "website_name": r.website_name, "store_id": r.store_id, "store_code": r.store_code, "store_name": r.store_name, "is_default": r.is_default, } for r in rows ] class SqlAlchemyCategoryRegistryRepository: """CRUD for magento_category_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def get_all(self, connection_id: int, ttl_cutoff: datetime) -> List[dict]: stmt = select(MagentoCategoryRegistry).where( MagentoCategoryRegistry.connection_id == connection_id, MagentoCategoryRegistry.fetched_at >= ttl_cutoff, ) return [ { "category_id": r.category_id, "parent_id": r.parent_id, "name": r.name, "path": r.path, "path_names": r.path_names, "level": r.level, "is_active": r.is_active, } for r in self._session.execute(stmt).scalars().all() ] def resolve_path(self, connection_id: int, path_names: str, ttl_cutoff: datetime) -> Optional[int]: norm = path_names.strip() if not norm: return None stmt = select(MagentoCategoryRegistry).where( MagentoCategoryRegistry.connection_id == connection_id, MagentoCategoryRegistry.path_names == norm, MagentoCategoryRegistry.fetched_at >= ttl_cutoff, ) row = self._session.execute(stmt).scalars().first() return row.category_id if row else None def resolve_name(self, connection_id: int, name: str, ttl_cutoff: datetime) -> Optional[int]: norm = name.strip().lower() if not norm: return None stmt = select(MagentoCategoryRegistry).where( MagentoCategoryRegistry.connection_id == connection_id, MagentoCategoryRegistry.is_active.is_(True), MagentoCategoryRegistry.fetched_at >= ttl_cutoff, ) for r in self._session.execute(stmt).scalars().all(): if r.name.strip().lower() == norm: return r.category_id return None def upsert_category(self, connection_id: int, category: dict, fetched_at: datetime) -> None: """Insert or update a single category (used after auto-creation via API).""" cat_id = category.get("category_id") if not cat_id: return stmt = select(MagentoCategoryRegistry).where( MagentoCategoryRegistry.connection_id == connection_id, MagentoCategoryRegistry.category_id == cat_id, ) row = self._session.execute(stmt).scalars().first() path_names = category.get("path_names") or "" if row: row.parent_id = category.get("parent_id") row.name = category.get("name", "") row.path = category.get("path", "") row.path_names = path_names row.level = category.get("level", 0) row.is_active = category.get("is_active", True) row.fetched_at = fetched_at else: self._session.add(MagentoCategoryRegistry( connection_id=connection_id, category_id=cat_id, parent_id=category.get("parent_id"), name=category.get("name", ""), path=category.get("path", ""), path_names=path_names, level=category.get("level", 0), is_active=category.get("is_active", True), fetched_at=fetched_at, )) self._session.flush() def bulk_replace(self, connection_id: int, categories: List[dict], fetched_at: datetime) -> int: """Upsert categories for a connection and deactivate ids missing from the pull. Magento category trees can list the same id more than once; also race with concurrent baseline pulls. Deduplicate input and upsert safely. """ # Last occurrence wins (tree walk order). deduped: Dict[int, dict] = {} for cat in categories: try: cat_id = int(cat["category_id"]) except (KeyError, TypeError, ValueError): continue deduped[cat_id] = cat if not deduped: return 0 bind = self._session.get_bind() if bind is not None and bind.dialect.name == "postgresql": rows = [ { "connection_id": connection_id, "category_id": cat_id, "parent_id": cat.get("parent_id"), "name": cat["name"], "path": cat.get("path") or "", "path_names": cat.get("path_names"), "level": cat.get("level", 0), "is_active": cat.get("is_active", True), "fetched_at": fetched_at, } for cat_id, cat in deduped.items() ] stmt = pg_insert(MagentoCategoryRegistry).values(rows) stmt = stmt.on_conflict_do_update( constraint="uq_category_registry_conn_catid", set_={ "parent_id": stmt.excluded.parent_id, "name": stmt.excluded.name, "path": stmt.excluded.path, "path_names": stmt.excluded.path_names, "level": stmt.excluded.level, "is_active": stmt.excluded.is_active, "fetched_at": stmt.excluded.fetched_at, }, ) self._session.execute(stmt) self._session.flush() self._deactivate_missing_categories(connection_id, set(deduped.keys())) return len(rows) existing_stmt = select(MagentoCategoryRegistry).where( MagentoCategoryRegistry.connection_id == connection_id, ) existing_map = { int(r.category_id): r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for cat_id, cat in deduped.items(): row = existing_map.get(cat_id) if row: row.parent_id = cat.get("parent_id") row.name = cat["name"] row.path = cat.get("path") or "" row.path_names = cat.get("path_names") row.level = cat.get("level", 0) row.is_active = cat.get("is_active", True) row.fetched_at = fetched_at else: row = MagentoCategoryRegistry( connection_id=connection_id, category_id=cat_id, parent_id=cat.get("parent_id"), name=cat["name"], path=cat.get("path") or "", path_names=cat.get("path_names"), level=cat.get("level", 0), is_active=cat.get("is_active", True), fetched_at=fetched_at, ) self._session.add(row) existing_map[cat_id] = row count += 1 self._session.flush() self._deactivate_missing_categories(connection_id, set(deduped.keys())) return count def _deactivate_missing_categories(self, connection_id: int, keep_ids: set[int]) -> int: """Mark registry rows inactive when Magento no longer returns them.""" rows = list( self._session.scalars( select(MagentoCategoryRegistry) .where(MagentoCategoryRegistry.connection_id == int(connection_id)) .where(MagentoCategoryRegistry.is_active.is_(True)) ).all() ) deactivated = 0 for row in rows: if int(row.category_id) in keep_ids: continue row.is_active = False deactivated += 1 if deactivated: self._session.flush() return deactivated class SqlAlchemyAttributeRegistryRepository: """CRUD for magento_attribute_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def get_attributes( self, connection_id: int, codes: List[str], ttl_cutoff: datetime ) -> Dict[str, dict]: if not codes: return {} codes_lower = [c.strip().lower() for c in codes if c.strip()] if not codes_lower: return {} stmt = select(MagentoAttributeRegistry).where( MagentoAttributeRegistry.connection_id == connection_id, MagentoAttributeRegistry.attribute_code.in_(codes_lower), MagentoAttributeRegistry.fetched_at >= ttl_cutoff, ) return { str(r.attribute_code).lower(): { "attribute_code": r.attribute_code, "attribute_id": r.attribute_id, "frontend_label": r.frontend_label, "frontend_input": r.frontend_input, "backend_type": r.backend_type, "is_user_defined": r.is_user_defined, "fetched_at": r.fetched_at, } for r in self._session.execute(stmt).scalars().all() } def upsert_attribute( self, connection_id: int, attribute_code: str, attribute_id: int, *, frontend_label: Optional[str] = None, frontend_input: Optional[str] = None, backend_type: Optional[str] = None, is_user_defined: Optional[bool] = None, fetched_at: datetime, ) -> None: code = attribute_code.strip().lower() if self._session.get_bind() is not None and self._session.get_bind().dialect.name == "postgresql": stmt = pg_insert(MagentoAttributeRegistry).values( connection_id=connection_id, attribute_code=code, attribute_id=attribute_id, frontend_label=frontend_label, frontend_input=frontend_input, backend_type=backend_type, is_user_defined=is_user_defined, fetched_at=fetched_at, ) stmt = stmt.on_conflict_do_update( constraint="uq_attribute_registry_conn_code", set_={ "attribute_id": stmt.excluded.attribute_id, "frontend_label": stmt.excluded.frontend_label, "frontend_input": stmt.excluded.frontend_input, "backend_type": stmt.excluded.backend_type, "is_user_defined": stmt.excluded.is_user_defined, "fetched_at": stmt.excluded.fetched_at, }, ) self._session.execute(stmt) self._session.flush() return stmt = select(MagentoAttributeRegistry).where( MagentoAttributeRegistry.connection_id == connection_id, MagentoAttributeRegistry.attribute_code == code, ) row = self._session.scalars(stmt).first() if row is None: row = MagentoAttributeRegistry( connection_id=connection_id, attribute_code=code, attribute_id=attribute_id, frontend_label=frontend_label, frontend_input=frontend_input, backend_type=backend_type, is_user_defined=is_user_defined, fetched_at=fetched_at, ) self._session.add(row) else: row.attribute_id = attribute_id row.frontend_label = frontend_label or row.frontend_label row.frontend_input = frontend_input or row.frontend_input row.backend_type = backend_type or row.backend_type row.is_user_defined = is_user_defined if is_user_defined is not None else row.is_user_defined row.fetched_at = fetched_at self._session.add(row) self._session.flush() def get_options( self, connection_id: int, attribute_code: str, ttl_cutoff: datetime ) -> Dict[str, int]: code = attribute_code.strip().lower() stmt = select(MagentoAttributeOptionRegistry).where( MagentoAttributeOptionRegistry.connection_id == connection_id, MagentoAttributeOptionRegistry.attribute_code == code, MagentoAttributeOptionRegistry.fetched_at >= ttl_cutoff, ) return { r.option_label_norm: r.value_index for r in self._session.execute(stmt).scalars().all() } def bulk_upsert_options( self, connection_id: int, attribute_code: str, options: List[Dict[str, Any]], fetched_at: datetime, ) -> None: code = attribute_code.strip().lower() for opt in options: label = str(opt.get("label", "")).strip() vi = opt.get("value_index") if opt.get("value_index") is not None else opt.get("value") if label and vi is not None: try: value_index = int(vi) except (ValueError, TypeError): continue norm = _norm_option_label(label) if not norm: continue stmt = select(MagentoAttributeOptionRegistry).where( MagentoAttributeOptionRegistry.connection_id == connection_id, MagentoAttributeOptionRegistry.attribute_code == code, MagentoAttributeOptionRegistry.option_label_norm == norm, ) row = self._session.scalars(stmt).first() if row is None: row = MagentoAttributeOptionRegistry( connection_id=connection_id, attribute_code=code, option_label=label, option_label_norm=norm, value_index=value_index, fetched_at=fetched_at, ) self._session.add(row) else: row.option_label = label row.value_index = value_index row.fetched_at = fetched_at self._session.add(row) self._session.flush() class SqlAlchemyAttributeSetRegistryRepository: """CRUD for magento_attribute_set_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace( self, connection_id: int, sets: List[Dict[str, Any]], fetched_at: datetime ) -> int: existing_stmt = select(MagentoAttributeSetRegistry).where( MagentoAttributeSetRegistry.connection_id == connection_id, ) existing_map = { r.attribute_set_id: r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for entry in sets: set_id = entry["attribute_set_id"] name = entry.get("attribute_set_name", "") row = existing_map.get(set_id) if row: row.attribute_set_name = name row.fetched_at = fetched_at else: row = MagentoAttributeSetRegistry( connection_id=connection_id, attribute_set_id=set_id, attribute_set_name=name, fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_by_name(self, connection_id: int, name: str) -> Optional[int]: norm = name.strip() if not norm: return None stmt = select(MagentoAttributeSetRegistry).where( MagentoAttributeSetRegistry.connection_id == connection_id, MagentoAttributeSetRegistry.attribute_set_name == norm, ) row = self._session.execute(stmt).scalars().first() return row.attribute_set_id if row else None def get_first_by_id(self, connection_id: int) -> Optional[int]: stmt = ( select(MagentoAttributeSetRegistry) .where(MagentoAttributeSetRegistry.connection_id == connection_id) .order_by(MagentoAttributeSetRegistry.attribute_set_id.asc()) .limit(1) ) row = self._session.execute(stmt).scalars().first() return row.attribute_set_id if row else None def get_by_id(self, connection_id: int, attribute_set_id: int) -> Optional[int]: """Return attribute_set_id if it exists in registry (valid for this connection), else None.""" stmt = select(MagentoAttributeSetRegistry).where( MagentoAttributeSetRegistry.connection_id == connection_id, MagentoAttributeSetRegistry.attribute_set_id == attribute_set_id, ) row = self._session.execute(stmt).scalars().first() return row.attribute_set_id if row else None class SqlAlchemyAttributeSetAttributesRepository: """CRUD for magento_attribute_set_attributes: set_id -> attribute_code, attribute_id.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace_for_set( self, connection_id: int, attribute_set_id: int, attributes: List[Dict[str, Any]], fetched_at: datetime, ) -> int: existing_stmt = select(MagentoAttributeSetAttributes).where( MagentoAttributeSetAttributes.connection_id == connection_id, MagentoAttributeSetAttributes.attribute_set_id == attribute_set_id, ) existing_map = { r.attribute_code: r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for att in attributes: code = str(att.get("attribute_code") or att.get("attributeCode", "")).strip().lower() attr_id = att.get("attribute_id") or att.get("attributeId") if not code or attr_id is None: continue row = existing_map.get(code) if row: row.attribute_id = int(attr_id) row.fetched_at = fetched_at else: row = MagentoAttributeSetAttributes( connection_id=connection_id, attribute_set_id=attribute_set_id, attribute_id=int(attr_id), attribute_code=code, fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_attributes_for_set( self, connection_id: int, attribute_set_id: int ) -> Dict[str, int]: """Return {attribute_code: attribute_id} for attributes in the set.""" stmt = select(MagentoAttributeSetAttributes).where( MagentoAttributeSetAttributes.connection_id == connection_id, MagentoAttributeSetAttributes.attribute_set_id == attribute_set_id, ) rows = self._session.execute(stmt).scalars().all() return {r.attribute_code: r.attribute_id for r in rows} def attribute_in_set( self, connection_id: int, attribute_set_id: int, attribute_code: str ) -> bool: code = attribute_code.strip().lower() stmt = select(MagentoAttributeSetAttributes).where( MagentoAttributeSetAttributes.connection_id == connection_id, MagentoAttributeSetAttributes.attribute_set_id == attribute_set_id, MagentoAttributeSetAttributes.attribute_code == code, ) return self._session.execute(stmt).scalars().first() is not None class SqlAlchemySourceRegistryRepository: """CRUD for magento_source_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace( self, connection_id: int, sources: List[Dict[str, Any]], fetched_at: datetime ) -> int: existing_stmt = select(MagentoSourceRegistry).where( MagentoSourceRegistry.connection_id == connection_id, ) existing_map = { r.source_code: r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for s in sources: code = str(s.get("source_code") or s.get("code", "")).strip() if not code: continue row = existing_map.get(code) name = s.get("name") enabled = s.get("enabled") country_id = s.get("country_id") postcode = s.get("postcode") if row: row.name = name row.enabled = enabled row.country_id = country_id row.postcode = postcode row.fetched_at = fetched_at else: row = MagentoSourceRegistry( connection_id=connection_id, source_code=code, name=name, enabled=enabled, country_id=country_id, postcode=postcode, fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_enabled_source_codes(self, connection_id: int) -> List[str]: """Return list of enabled source codes for the connection.""" stmt = select(MagentoSourceRegistry).where( MagentoSourceRegistry.connection_id == connection_id, MagentoSourceRegistry.enabled.is_(True), ).order_by(MagentoSourceRegistry.source_code) rows = self._session.execute(stmt).scalars().all() return [r.source_code for r in rows] def get_all_source_codes(self, connection_id: int) -> List[str]: """Return all source codes (enabled or not) for the connection.""" stmt = select(MagentoSourceRegistry).where( MagentoSourceRegistry.connection_id == connection_id, ).order_by(MagentoSourceRegistry.source_code) rows = self._session.execute(stmt).scalars().all() return [r.source_code for r in rows] class SqlAlchemyStockRegistryRepository: """CRUD for magento_stock_registry populated by baseline sync.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace( self, connection_id: int, stocks: List[Dict[str, Any]], fetched_at: datetime ) -> int: existing_stmt = select(MagentoStockRegistry).where( MagentoStockRegistry.connection_id == connection_id, ) existing_map = { r.stock_id: r for r in self._session.execute(existing_stmt).scalars().all() } count = 0 for s in stocks: stock_id = s.get("stock_id") if stock_id is None: continue stock_id = int(stock_id) ext = s.get("extension_attributes") or {} sales_channels = ext.get("sales_channels") or [] row = existing_map.get(stock_id) if row: row.name = str(s.get("name", "")) row.sales_channels_json = sales_channels if sales_channels else None row.fetched_at = fetched_at else: row = MagentoStockRegistry( connection_id=connection_id, stock_id=stock_id, name=str(s.get("name", "")), sales_channels_json=sales_channels if sales_channels else None, fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_stock_ids(self, connection_id: int) -> List[int]: """Return list of stock IDs for the connection.""" stmt = select(MagentoStockRegistry).where( MagentoStockRegistry.connection_id == connection_id, ).order_by(MagentoStockRegistry.stock_id) rows = self._session.execute(stmt).scalars().all() return [r.stock_id for r in rows] class SqlAlchemyStockSourceLinksRepository: """CRUD for magento_stock_source_links: stock_id -> source_code assignments.""" def __init__(self, session: Session) -> None: self._session = session def bulk_replace_for_connection( self, connection_id: int, links: List[Dict[str, Any]], fetched_at: datetime, ) -> int: """Replace all stock-source links for this connection.""" delete_stmt = select(MagentoStockSourceLinks).where( MagentoStockSourceLinks.connection_id == connection_id, ) for row in self._session.execute(delete_stmt).scalars().all(): self._session.delete(row) self._session.flush() count = 0 for link in links: stock_id = link.get("stock_id") source_code = str(link.get("source_code", "")).strip() if stock_id is None or not source_code: continue row = MagentoStockSourceLinks( connection_id=connection_id, stock_id=int(stock_id), source_code=source_code, priority=link.get("priority"), fetched_at=fetched_at, ) self._session.add(row) count += 1 self._session.flush() return count def get_sources_for_stock( self, connection_id: int, stock_id: int ) -> List[str]: """Return source codes assigned to the stock, ordered by priority.""" stmt = select(MagentoStockSourceLinks).where( MagentoStockSourceLinks.connection_id == connection_id, MagentoStockSourceLinks.stock_id == stock_id, ).order_by( MagentoStockSourceLinks.priority.desc().nulls_last(), MagentoStockSourceLinks.source_code, ) rows = self._session.execute(stmt).scalars().all() return [r.source_code for r in rows] def get_all_sources_for_stocks( self, connection_id: int, stock_ids: Optional[List[int]] = None ) -> List[str]: """Return unique source codes for given stocks (or all stocks).""" stmt = select(MagentoStockSourceLinks.source_code).where( MagentoStockSourceLinks.connection_id == connection_id, ).distinct() if stock_ids: stmt = stmt.where(MagentoStockSourceLinks.stock_id.in_(stock_ids)) rows = self._session.execute(stmt).scalars().all() return list(dict.fromkeys(r.source_code for r in rows)) class SqlAlchemyBaselineSyncRunRepository: """Track baseline sync run status for freshness gating.""" def __init__(self, session: Session) -> None: self._session = session def create_run(self, connection_id: int, started_at: datetime) -> int: row = MagentoBaselineSyncRun( connection_id=connection_id, started_at=started_at, status="running", ) self._session.add(row) self._session.flush() return int(row.id) def mark_success(self, run_id: int, fetched_counts: Dict[str, Any]) -> None: row = self._session.get(MagentoBaselineSyncRun, run_id) if row: row.finished_at = datetime.utcnow() row.status = "success" row.fetched_counts = fetched_counts self._session.add(row) self._session.flush() def mark_failed(self, run_id: int, error: str) -> None: row = self._session.get(MagentoBaselineSyncRun, run_id) if row: row.finished_at = datetime.utcnow() row.status = "failed" row.error = (error or "")[:2000] self._session.add(row) self._session.flush() def get_last_successful(self, connection_id: int) -> Optional[Dict[str, Any]]: stmt = ( select(MagentoBaselineSyncRun) .where( MagentoBaselineSyncRun.connection_id == connection_id, MagentoBaselineSyncRun.status == "success", ) .order_by(MagentoBaselineSyncRun.finished_at.desc()) .limit(1) ) row = self._session.execute(stmt).scalars().first() if not row: return None return { "id": row.id, "connection_id": row.connection_id, "started_at": row.started_at, "finished_at": row.finished_at, "fetched_counts": row.fetched_counts, } def get_default_attribute_set_id(self, connection_id: int) -> Optional[int]: """From last successful run's fetched_counts, return default_attribute_set_id.""" last = self.get_last_successful(connection_id) if not last or not last.get("fetched_counts"): return None fc = last["fetched_counts"] if not isinstance(fc, dict): return None val = fc.get("default_attribute_set_id") if val is None: return None try: return int(val) except (TypeError, ValueError): return None def is_fresh(self, connection_id: int, ttl_hours: int) -> bool: last = self.get_last_successful(connection_id) if not last or not last.get("finished_at"): return False cutoff = datetime.utcnow() - timedelta(hours=ttl_hours) finished = last["finished_at"] if hasattr(finished, "replace"): finished = finished.replace(tzinfo=None) return finished >= cutoff class SqlAlchemyMagentoProductOverrideRepository: """Load magento_product_override rows for sync. Batch query by connection+skus.""" def __init__(self, session: Session) -> None: self._session = session def get_overrides_for_skus( self, connection_id: int, skus: List[str] ) -> OverrideSet: """ Batch load overrides. Returns OverrideSet with force (value) and lock (exclude) per sku. - sync_lock_reason NOT NULL -> lock (exclude from payload) - sync_lock_reason NULL/empty -> force (set override value) """ if not skus: return OverrideSet(force={}, lock={}) stmt = select(MagentoProductOverride).where( MagentoProductOverride.connection_id == connection_id, MagentoProductOverride.sku.in_(skus), ) rows = self._session.execute(stmt).scalars().all() force: dict = {} lock: dict = {} for row in rows: sku = row.sku field = row.field is_lock = bool(row.sync_lock_reason and str(row.sync_lock_reason).strip()) if is_lock: lock.setdefault(sku, set()).add(field) else: force.setdefault(sku, {})[field] = row.value return OverrideSet(force=force, lock=lock) class SqlAlchemyMagentoCatalogStateRepository: """Upsert pulled Magento product state (price, media, etc).""" def __init__(self, session: Session) -> None: self._session = session def upsert_catalog_state( self, connection_id: int, sku: str, *, magento_product_id: Optional[int] = None, updated_at_magento: Optional[datetime] = None, price: Optional[float] = None, special_price: Optional[float] = None, media_files: Optional[list] = None, media_hash: Optional[str] = None, payload_hash: Optional[str] = None, ) -> None: stmt = select(MagentoCatalogState).where( MagentoCatalogState.connection_id == connection_id, MagentoCatalogState.sku == sku, ) row = self._session.execute(stmt).scalars().first() now = datetime.utcnow() if row is None: row = MagentoCatalogState( connection_id=connection_id, sku=sku, magento_product_id=magento_product_id, updated_at_magento=updated_at_magento, price=price, special_price=special_price, media_files=media_files, media_hash=media_hash, payload_hash=payload_hash, last_pulled_at=now, ) self._session.add(row) else: if magento_product_id is not None: row.magento_product_id = magento_product_id if updated_at_magento is not None: row.updated_at_magento = updated_at_magento if price is not None: row.price = price if special_price is not None: row.special_price = special_price if media_files is not None: row.media_files = media_files if media_hash is not None: row.media_hash = media_hash if payload_hash is not None: row.payload_hash = payload_hash row.last_pulled_at = now self._session.add(row) self._session.flush() def get_existing_magento_skus(self, connection_id: int, skus: List[str]) -> set: """Return set of SKUs that have a magento_product_id (confirmed to exist in Magento).""" if not skus: return set() stmt = select(MagentoCatalogState.sku).where( MagentoCatalogState.connection_id == connection_id, MagentoCatalogState.sku.in_(skus), MagentoCatalogState.magento_product_id.isnot(None), ) return {row for (row,) in self._session.execute(stmt).all()} def get_catalog_state_for_skus( self, connection_id: int, skus: List[str] ) -> Dict[str, Dict[str, Any]]: """Return {sku: {price, magento_product_id}} for planner price/status logic.""" if not skus: return {} stmt = select(MagentoCatalogState).where( MagentoCatalogState.connection_id == connection_id, MagentoCatalogState.sku.in_(skus), ) rows = self._session.execute(stmt).scalars().all() return { r.sku: { "price": float(r.price) if r.price is not None else None, "magento_product_id": r.magento_product_id, } for r in rows } def reset_magento_ids( self, connection_id: int, skus: Optional[List[str]] = None, ) -> int: """Clear stale Magento IDs and media state after a product purge. Preserves price and special_price (used to re-push correct prices on re-create). If skus is provided, only those SKUs are reset; otherwise all for the connection. Returns the number of rows updated.""" stmt = ( update(MagentoCatalogState) .where(MagentoCatalogState.connection_id == connection_id) .values( magento_product_id=None, updated_at_magento=None, last_pulled_at=None, media_files=None, media_hash=None, payload_hash=None, ) ) if skus: stmt = stmt.where(MagentoCatalogState.sku.in_(skus)) result = self._session.execute(stmt) return result.rowcount def get_media_hashes_for_skus( self, connection_id: int, skus: List[str] ) -> Dict[str, Optional[str]]: """Return {sku: media_hash}. Used by planner to know if Magento already has images.""" if not skus: return {} stmt = select(MagentoCatalogState).where( MagentoCatalogState.connection_id == connection_id, MagentoCatalogState.sku.in_(skus), MagentoCatalogState.media_hash.isnot(None), ) rows = self._session.execute(stmt).scalars().all() return {r.sku: r.media_hash for r in rows} class SqlAlchemyMagentoSyncQueueRepository: """Enqueue and process magento_sync_queue.""" def __init__(self, session: Session) -> None: self._session = session def enqueue( self, connection_id: int, label: str, options: Optional[dict] = None, *, supersede_queued: bool = False, ) -> int: if supersede_queued: self.cancel_queued( connection_id=connection_id, reason=f"superseded by newer queue label={label}", ) row = MagentoSyncQueue( connection_id=connection_id, label=label, status="queued", options=options, ) self._session.add(row) self._session.flush() return int(row.id) def claim_by_id(self, queue_id: int) -> Optional[dict]: """Atomically claim a specific queued item (for targeted sample/CLI runs).""" from datetime import datetime stmt = ( select(MagentoSyncQueue) .where( MagentoSyncQueue.id == queue_id, MagentoSyncQueue.status == "queued", MagentoSyncQueue.attempts < MagentoSyncQueue.max_attempts, ) .with_for_update(skip_locked=True) ) row = self._session.execute(stmt).scalars().first() if not row: return None row.status = "running" row.started_at = datetime.utcnow() row.attempts = (row.attempts or 0) + 1 self._session.flush() return { "id": row.id, "connection_id": row.connection_id, "label": row.label, "options": row.options or {}, "attempt": row.attempts, "job_id": int(row.job_id) if row.job_id is not None else None, "queued_at": row.queued_at, } def claim_next(self) -> Optional[dict]: """Atomically claim the next queued item using SELECT FOR UPDATE SKIP LOCKED. Combines get_next_queued + mark_running in one transaction-safe operation, preventing two concurrent workers from claiming the same item. Only claims items that haven't exceeded max_attempts. """ from datetime import datetime stmt = ( select(MagentoSyncQueue) .where( MagentoSyncQueue.status == "queued", MagentoSyncQueue.attempts < MagentoSyncQueue.max_attempts, ) .order_by(MagentoSyncQueue.queued_at) .limit(1) .with_for_update(skip_locked=True) ) row = self._session.execute(stmt).scalars().first() if not row: return None row.status = "running" row.started_at = datetime.utcnow() row.attempts = (row.attempts or 0) + 1 self._session.flush() return { "id": row.id, "connection_id": row.connection_id, "label": row.label, "options": row.options or {}, "attempt": row.attempts, "job_id": int(row.job_id) if row.job_id is not None else None, "queued_at": row.queued_at, } def reset_stale_running(self, stale_minutes: int = 120) -> int: """Reset items stuck in 'running' back to 'queued' for retry. Called at worker startup and periodically to recover from crashes. Items that have exhausted max_attempts are marked 'failed' instead. Returns the number of items reset. """ from datetime import datetime, timedelta cutoff = datetime.utcnow() - timedelta(minutes=stale_minutes) stmt = ( select(MagentoSyncQueue) .where( MagentoSyncQueue.status == "running", MagentoSyncQueue.started_at < cutoff, ) .with_for_update(skip_locked=True) ) rows = self._session.execute(stmt).scalars().all() reset_count = 0 for row in rows: if row.attempts >= row.max_attempts: row.status = "failed" row.finished_at = datetime.utcnow() row.last_error = f"Exhausted {row.max_attempts} attempts (last stale reset)" else: row.status = "queued" row.started_at = None reset_count += 1 if reset_count: self._session.flush() return reset_count # Keep for backwards compatibility def get_next_queued(self) -> Optional[dict]: return self.claim_next() def mark_running(self, queue_id: int) -> None: """No-op: claim_next() already marks running atomically.""" pass def has_scheduled_sync_for_date(self, for_date) -> bool: """True if any queue item with label 'scheduled-YYYYMMDD-*' exists for the given date. Used to avoid re-enqueueing scheduled sync on app restart (in-memory _last_scheduled_sync_date resets).""" prefix = f"scheduled-{for_date.strftime('%Y%m%d')}-" stmt = ( select(MagentoSyncQueue.id) .where(MagentoSyncQueue.label.like(f"{prefix}%")) .limit(1) ) return self._session.execute(stmt).first() is not None def cancel_queued(self, *, connection_id: Optional[int] = None, reason: str = "cancelled by operator") -> int: """Mark queued (not running) sync items as cancelled. Returns count cancelled.""" from datetime import datetime stmt = select(MagentoSyncQueue).where(MagentoSyncQueue.status == "queued") if connection_id is not None: stmt = stmt.where(MagentoSyncQueue.connection_id == connection_id) rows = self._session.execute(stmt).scalars().all() for row in rows: row.status = "cancelled" row.finished_at = datetime.utcnow() row.last_error = reason[:2000] if rows: self._session.flush() return len(rows) def has_newer_active_item(self, queue_id: int, connection_id: int, queued_at) -> bool: """True when a newer queued/running item exists for the same connection.""" stmt = ( select(MagentoSyncQueue.id) .where( MagentoSyncQueue.connection_id == connection_id, MagentoSyncQueue.id != queue_id, MagentoSyncQueue.status.in_(["queued", "running"]), MagentoSyncQueue.queued_at > queued_at, ) .limit(1) ) return self._session.execute(stmt).first() is not None def set_job_id(self, queue_id: int, job_id: int) -> None: """Link a sync job to the queue item for resumable sync.""" row = self._session.get(MagentoSyncQueue, queue_id) if row: row.job_id = job_id self._session.add(row) def mark_done(self, queue_id: int) -> None: row = self._session.get(MagentoSyncQueue, queue_id) if row: from datetime import datetime row.status = "done" row.finished_at = datetime.utcnow() row.last_error = None self._session.add(row) def mark_cancelled(self, queue_id: int, reason: str = "cancelled") -> None: row = self._session.get(MagentoSyncQueue, queue_id) if row: from datetime import datetime row.status = "cancelled" row.finished_at = datetime.utcnow() row.last_error = (reason or "cancelled")[:2000] self._session.add(row) def mark_failed(self, queue_id: int, error: str) -> None: row = self._session.get(MagentoSyncQueue, queue_id) if row: from datetime import datetime row.status = "failed" row.finished_at = datetime.utcnow() row.last_error = (error or "")[:2000] self._session.add(row) def get_status(self, queue_id: int) -> Optional[dict]: """Queue row plus linked sync-job progress for dashboard polling.""" row = self._session.get(MagentoSyncQueue, queue_id) if row is None: return None def _dt(value: Optional[datetime]) -> Optional[str]: return value.isoformat() if value is not None else None progress: dict = { "total": None, "completed": 0, "success_count": 0, "error_count": 0, "percent": None, "job_status": None, } if row.job_id: job = self._session.get(MagentoSyncJob, row.job_id) if job: total = job.total_count success = job.success_count or 0 errors = job.error_count or 0 completed = success + errors progress = { "total": total, "completed": completed, "success_count": success, "error_count": errors, "percent": round(100 * completed / total, 1) if total and total > 0 else None, "job_status": job.status, } terminal = row.status in {"done", "failed"} return { "id": row.id, "connection_id": row.connection_id, "label": row.label, "status": row.status, "queued_at": _dt(row.queued_at), "started_at": _dt(row.started_at), "finished_at": _dt(row.finished_at), "last_error": row.last_error, "job_id": row.job_id, "options": row.options or {}, "progress": progress, "terminal": terminal, } class SqlAlchemyPlytixFeedEventRepository: def __init__(self, session: Session) -> None: self._session = session def create(self, label: str, type: str, payload: Optional[dict] = None) -> int: row = PlytixFeedEvent(label=label, type=type, payload=payload) self._session.add(row) self._session.flush() return int(row.id) class SqlAlchemyMagentoSyncNotificationRepository: """Enqueue and manage sync notification emails.""" def __init__(self, session: Session) -> None: self._session = session def enqueue_email( self, connection_id: int, job_id: int, to_email: str, subject: str, body: str ) -> int: row = MagentoSyncNotification( connection_id=connection_id, job_id=job_id, channel="email", to_email=to_email, subject=subject, body=body, status="queued", ) self._session.add(row) self._session.flush() return int(row.id) def mark_sent(self, notification_id: int) -> None: row = self._session.get(MagentoSyncNotification, notification_id) if row: row.status = "sent" row.sent_at = datetime.utcnow() self._session.add(row) def mark_failed(self, notification_id: int, error: str) -> None: row = self._session.get(MagentoSyncNotification, notification_id) if row: row.status = "failed" row.error = error[:1000] self._session.add(row) class SqlAlchemyMagentoProductVersionRepository: """Persist and query product version snapshots for rollback.""" def __init__(self, session: Session) -> None: self._session = session def create_version( self, connection_id: int, sku: str, *, job_id: Optional[int] = None, source: str = "sync", data_hash: Optional[str] = None, images_hash: Optional[str] = None, relations_hash: Optional[str] = None, product_payload_json: Optional[dict] = None, media_payload_json: Optional[dict] = None, relations_payload_json: Optional[dict] = None, status: str = "planned", ) -> int: row = MagentoProductVersion( connection_id=connection_id, sku=sku, job_id=job_id, source=source, data_hash=data_hash, images_hash=images_hash, relations_hash=relations_hash, product_payload_json=product_payload_json, media_payload_json=media_payload_json, relations_payload_json=relations_payload_json, status=status, ) self._session.add(row) self._session.flush() return int(row.id) def update_version_status(self, version_id: int, status: str) -> None: row = self._session.get(MagentoProductVersion, version_id) if not row: return row.status = status self._session.add(row) def get_version_by_id(self, version_id: int) -> Optional[dict]: row = self._session.get(MagentoProductVersion, version_id) if not row or row.status != "applied": return None return { "id": row.id, "connection_id": row.connection_id, "sku": row.sku, "job_id": row.job_id, "created_at": row.created_at, "source": row.source, "status": row.status, "data_hash": row.data_hash, "images_hash": row.images_hash, "relations_hash": row.relations_hash, "product_payload_json": row.product_payload_json, "media_payload_json": row.media_payload_json, "relations_payload_json": row.relations_payload_json, } def get_latest_versions_for_skus( self, connection_id: int, skus: List[str] ) -> Dict[str, dict]: """Return {sku: version_dict} for the latest version of each sku.""" if not skus: return {} from sqlalchemy import func subq = ( select( MagentoProductVersion.sku, func.max(MagentoProductVersion.created_at).label("max_created"), ) .where( MagentoProductVersion.connection_id == connection_id, MagentoProductVersion.sku.in_(skus), MagentoProductVersion.status == "applied", ) .group_by(MagentoProductVersion.sku) ) subq = subq.subquery() stmt = ( select(MagentoProductVersion) .join( subq, (MagentoProductVersion.sku == subq.c.sku) & (MagentoProductVersion.created_at == subq.c.max_created) & (MagentoProductVersion.connection_id == connection_id), ) .where(MagentoProductVersion.connection_id == connection_id) ) rows = self._session.execute(stmt).scalars().all() return { r.sku: { "id": r.id, "connection_id": r.connection_id, "sku": r.sku, "job_id": r.job_id, "created_at": r.created_at, "source": r.source, "status": r.status, "data_hash": r.data_hash, "images_hash": r.images_hash, "relations_hash": r.relations_hash, "product_payload_json": r.product_payload_json, "media_payload_json": r.media_payload_json, "relations_payload_json": r.relations_payload_json, } for r in rows } def get_versions_as_of( self, connection_id: int, as_of: datetime, sku: Optional[str] = None ) -> List[dict]: """Get latest version per SKU with created_at <= as_of.""" from sqlalchemy import func conditions = [ MagentoProductVersion.connection_id == connection_id, MagentoProductVersion.created_at <= as_of, MagentoProductVersion.status == "applied", ] if sku: conditions.append(MagentoProductVersion.sku == sku) subq = ( select( MagentoProductVersion.sku, func.max(MagentoProductVersion.created_at).label("max_created"), ) .where(*conditions) .group_by(MagentoProductVersion.sku) ).subquery() stmt = ( select(MagentoProductVersion) .join( subq, (MagentoProductVersion.sku == subq.c.sku) & (MagentoProductVersion.created_at == subq.c.max_created), ) .where( MagentoProductVersion.connection_id == connection_id, ) ) rows = self._session.execute(stmt).scalars().all() return [ { "id": r.id, "connection_id": r.connection_id, "sku": r.sku, "job_id": r.job_id, "created_at": r.created_at, "source": r.source, "status": r.status, "data_hash": r.data_hash, "images_hash": r.images_hash, "relations_hash": r.relations_hash, "product_payload_json": r.product_payload_json, "media_payload_json": r.media_payload_json, "relations_payload_json": r.relations_payload_json, } for r in rows ]