""" Process magento_sync_queue: run incremental sync for queued labels. CLI: python -m app.jobs.magento_sync_worker [--once] Debug: Set MAGENTO_SYNC_DEBUG=1 for verbose logging. Set MAGENTO_SYNC_TRACE_SKUS=SKU1,SKU2 to trace specific SKUs through pipeline. """ from __future__ import annotations import argparse import logging import os import sys import time from datetime import date, datetime, timedelta, timezone from typing import Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Debug controls: MAGENTO_SYNC_DEBUG=1 enables verbose logs; MAGENTO_SYNC_TRACE_SKUS=SKU1,SKU2 traces those SKUs _DEBUG = os.getenv("MAGENTO_SYNC_DEBUG", "").strip() in ("1", "true", "yes") _TRACE_SKUS: set[str] = set() _raw = os.getenv("MAGENTO_SYNC_TRACE_SKUS", "").strip() if _raw: _TRACE_SKUS = {s.strip() for s in _raw.split(",") if s.strip()} def _trace_sku(sku: str, trace_set: Optional[set[str]] = None) -> bool: """True if this SKU should be traced (case-insensitive match). Uses trace_set or _TRACE_SKUS.""" s = trace_set if trace_set is not None else _TRACE_SKUS return bool(s and sku and sku.upper() in {x.upper() for x in s}) def _persist_magento_sync_job_progress( job_id: int, *, total_count: int, success_count: int, error_count: int, ) -> None: """Update dashboard counters on a short-lived session. Progress must not share the long-lived sync session: committing that session while also locking ``magento_sync_jobs`` (after a large pending-plan flush) has deadlocked with concurrent readers/writers. """ import time from sqlalchemy.exc import OperationalError from db.models import MagentoSyncJob from db.session import get_session attempts = 3 for attempt in range(1, attempts + 1): try: with get_session() as prog_session: row = prog_session.get(MagentoSyncJob, job_id) if row is None: return row.total_count = int(total_count) row.success_count = int(success_count) row.error_count = int(error_count) prog_session.commit() return except OperationalError as exc: msg = str(exc).lower() if "deadlock" not in msg or attempt >= attempts: raise time.sleep(0.05 * attempt) def _filter_pending_deletion_review( session, connection_id: int, snapshot_rows: list, queue_id ) -> list: """ Remove SKUs that have a pending deletion review from the snapshot rows to be synced. This 'holds' those SKUs — no Magento API calls will be made for them until the deletion review is decided (approved or rejected). """ if not snapshot_rows: return snapshot_rows try: from db.models import MagentoDeletionReview from sqlalchemy import select all_skus = [str(row.get("sku", "")).strip() for row in snapshot_rows if row.get("sku")] if not all_skus: return snapshot_rows stmt = ( select(MagentoDeletionReview.sku) .where( MagentoDeletionReview.connection_id == connection_id, MagentoDeletionReview.status == "pending", MagentoDeletionReview.sku.in_(all_skus), ) ) pending_skus: set[str] = {row for (row,) in session.execute(stmt).all()} if not pending_skus: return snapshot_rows filtered = [ row for row in snapshot_rows if str(row.get("sku", "")).strip() not in pending_skus ] held_count = len(snapshot_rows) - len(filtered) logger.info( "[queue=%s] deletion_review_hold: held %d SKU(s) with pending deletion review: %s", queue_id, held_count, sorted(pending_skus), ) return filtered except Exception as exc: logger.warning( "[queue=%s] deletion_review_hold: error checking pending reviews (skipping hold): %s", queue_id, exc, ) return snapshot_rows def run_one(queue_id: Optional[int] = None) -> bool: """Process one queued item. Returns True if processed, False if none. When ``queue_id`` is set, only that row is claimed (if still queued). """ from db.session import get_session from db.magento_repositories import ( SqlAlchemyAttributeRegistryRepository, SqlAlchemyAttributeSetAttributesRepository, SqlAlchemyAttributeSetRegistryRepository, SqlAlchemyBaselineSyncRunRepository, SqlAlchemyCategoryRegistryRepository, SqlAlchemyMagentoConnectionRepository, SqlAlchemyMagentoSyncRepository, SqlAlchemyMagentoSyncPendingRepository, SqlAlchemyMagentoSyncStateRepository, SqlAlchemyMagentoMediaMapRepository, SqlAlchemyMagentoProductOverrideRepository, SqlAlchemyMagentoProductVersionRepository, SqlAlchemyMagentoSyncNotificationRepository, SqlAlchemyMagentoSyncQueueRepository, SqlAlchemyMagentoCatalogStateRepository, SqlAlchemyMagentoStoreRegistryRepository, SqlAlchemySourceRegistryRepository, ) from db.repositories import SqlAlchemyIngestRepository from db.url_shortener import resolve_slug from settings import load_magento_config, load_db_config, load_attribute_registry_codes, load_attribute_registry_options from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from magento.magento_api import MagentoRestClient from magento.sync_service import MagentoSyncService from db.schemas import MagentoSyncJobCreate with get_session() as session: queue_repo = SqlAlchemyMagentoSyncQueueRepository(session) # Recover any items left stuck in 'running' from a previous crash recovered = queue_repo.reset_stale_running(stale_minutes=120) if recovered: logger.warning("run_one: reset %d stale running item(s) back to queued", recovered) session.commit() item = queue_repo.claim_by_id(queue_id) if queue_id is not None else queue_repo.claim_next() if not item: return False session.commit() # commit the claim so other workers see it immediately queue_id = item["id"] connection_id = item["connection_id"] label = item["label"] options = item.get("options") or {} dry_run = options.get("dry_run", False) limit_skus = options.get("limit_skus") force_relations = options.get("force_relations", False) replace_options = options.get("replace_options", False) force_full_sync = options.get("force_full_sync", False) images_only = bool(options.get("images_only", False)) products_only = bool(options.get("products_only", False)) selected_attribute_codes = options.get("selected_attribute_codes") sync_options_from_master_enabled = bool(options.get("sync_options_from_master", True)) allow_parallel_with_newer_queue_items = bool( options.get("allow_parallel_with_newer_queue_items", False) ) use_connection_pending_table = bool( options.get("use_connection_pending_table", True) ) force_images = bool(options.get("force_images", False)) force_upsert = bool(options.get("force_upsert") or options.get("force_products")) force_upsert_skus = options.get("force_upsert_skus") if force_upsert and not force_upsert_skus and limit_skus: force_upsert_skus = list(limit_skus) elif force_upsert and not force_upsert_skus: force_upsert_skus = None # sync_incremental will need explicit list; set after snapshots expand_relations = options.get("expand_relations") if expand_relations is None: expand_relations = limit_skus is None from magento.sync_defaults import magento_sync_uses_master_catalog magento_sync_uses_master_catalog(options) debug_output_path = options.get("debug_output_path") options_trace_skus = options.get("trace_skus") # from API; overrides env run_pull_first = options.get("run_pull_first", True) attempt = item.get("attempt", 1) queued_at = item.get("queued_at") # Effective trace SKUs: API options override env effective_trace_skus = ( {str(s).strip() for s in options_trace_skus if s} if options_trace_skus else _TRACE_SKUS ) logger.info("[queue=%s] claimed attempt=%s connection=%s label=%s", queue_id, attempt, connection_id, label) if ( queued_at and not allow_parallel_with_newer_queue_items and queue_repo.has_newer_active_item(queue_id, connection_id, queued_at) ): message = "Cancelled before execution because a newer Magento queue item exists for this connection" logger.warning("[queue=%s] %s", queue_id, message) queue_repo.mark_cancelled(queue_id, message) session.commit() return True try: conn_repo = SqlAlchemyMagentoConnectionRepository(session) sync_repo = SqlAlchemyMagentoSyncRepository(session) pending_repo = SqlAlchemyMagentoSyncPendingRepository(session) state_repo = SqlAlchemyMagentoSyncStateRepository(session) media_repo = SqlAlchemyMagentoMediaMapRepository(session) override_repo = SqlAlchemyMagentoProductOverrideRepository(session) version_repo = SqlAlchemyMagentoProductVersionRepository(session) notification_repo = SqlAlchemyMagentoSyncNotificationRepository(session) ingest_repo = SqlAlchemyIngestRepository(session) catalog_repo = SqlAlchemyMagentoCatalogStateRepository(session) attr_registry_repo = SqlAlchemyAttributeRegistryRepository(session) attr_set_attrs_repo = SqlAlchemyAttributeSetAttributesRepository(session) attr_set_registry_repo = SqlAlchemyAttributeSetRegistryRepository(session) cat_registry_repo = SqlAlchemyCategoryRegistryRepository(session) baseline_repo = SqlAlchemyBaselineSyncRunRepository(session) store_registry_repo = SqlAlchemyMagentoStoreRegistryRepository(session) source_repo = SqlAlchemySourceRegistryRepository(session) def _source_codes_provider(cid: int): codes = source_repo.get_enabled_source_codes(cid) if not codes: codes = source_repo.get_all_source_codes(cid) if not codes: import os configured = os.getenv("MAGENTO_DEFAULT_SOURCE_CODE", "").strip() if configured: return [configured] return codes or ["default"] conn = conn_repo.get_for_sync(connection_id) if not conn: raise RuntimeError(f"Connection {connection_id} not found") # ---- Baseline freshness gate ---- from settings import load_magento_baseline_config baseline_cfg = load_magento_baseline_config() if images_only or products_only: logger.info( "[queue=%s] %s: skipping baseline freshness gate", queue_id, "images_only" if images_only else "products_only", ) elif not baseline_repo.is_fresh(connection_id, baseline_cfg.ttl_hours): logger.info("[queue=%s] Baseline stale/missing; running baseline sync first", queue_id) from app.jobs.magento_baseline_sync import run_baseline_sync bl_result = run_baseline_sync(connection_id, force=True, session=session) session.commit() if bl_result.get("status") == "failed": raise RuntimeError( f"Baseline sync failed, cannot proceed with product sync: {bl_result.get('error')}" ) logger.info("[queue=%s] Baseline sync completed: %s", queue_id, bl_result.get("counts")) else: logger.info("[queue=%s] Baseline is fresh, proceeding with product sync", queue_id) if not dry_run and not images_only: from db.channel_attribute_provision import provision_channel_attributes prov = provision_channel_attributes( session, "magento", connection_id=connection_id, dry_run=False, ) if prov.get("created") or prov.get("errors"): logger.info("[queue=%s] attribute_provision: %s", queue_id, prov) do_pull = run_pull_first if run_pull_first: import os env_skip = os.getenv("MAGENTO_RUN_PULL_BEFORE_SYNC", "true").strip().lower() in ("0", "false", "no") if env_skip: do_pull = False logger.info("[queue=%s] stage=pull: skipped (MAGENTO_RUN_PULL_BEFORE_SYNC=false)", queue_id) if do_pull and not images_only: logger.info("[queue=%s] stage=pull: pulling latest Magento state for connection=%s", queue_id, connection_id) from app.jobs.magento_pull import run_pull from settings import load_magento_pull_config cfg = load_magento_pull_config() since = date.today() - timedelta(days=cfg.lookback_days) run_pull(connection_id, since) session.commit() logger.info("[queue=%s] stage=pull: done", queue_id) elif do_pull and images_only: logger.info("[queue=%s] stage=pull: skipped (images_only)", queue_id) logger.info("[queue=%s] stage=load_skus source=master mode=%s", queue_id, "images_only" if images_only else "full") from db.channel_exports import resolve_push_skus sku_list = sorted( resolve_push_skus( session, "magento", skus=limit_skus, # Explicit limit_skus from dashboard/API wins over channel-assignment gate. only_assigned=limit_skus is None, expand_relations=expand_relations, ) ) from magento.sku_policy import filter_magento_push_skus filtered_sku_list = filter_magento_push_skus(sku_list) if len(filtered_sku_list) != len(sku_list): logger.info( "[queue=%s] stage=load_skus: excluded %d RTA SKU(s) from Magento push policy", queue_id, len(sku_list) - len(filtered_sku_list), ) sku_list = sorted(filtered_sku_list) if not sku_list: logger.info("[queue=%s] stage=load_skus: no Magento-assigned master SKUs, marking done", queue_id) from app.jobs.channel_jobs import report_channel_image_push_outcome if images_only: report_channel_image_push_outcome( session, queue_id, message="No Magento-assigned master SKUs. Assign SKUs on the Assignments page first.", ) queue_repo.mark_done(queue_id) session.commit() return True logger.info("[queue=%s] stage=load_skus: %d master-assigned SKUs", queue_id, len(sku_list)) if images_only: from magento.master_catalog_rows import build_magento_image_rows_from_master snapshot_rows = build_magento_image_rows_from_master( session, skus=sku_list, only_assigned=False, connection_id=connection_id, ) logger.info("[queue=%s] stage=snapshots: %d image-only master rows built", queue_id, len(snapshot_rows)) assigned_count = len(sku_list) if not snapshot_rows and images_only: from app.jobs.channel_jobs import report_channel_image_push_outcome report_channel_image_push_outcome( session, queue_id, assigned_sku_count=assigned_count, skus_with_images=0, planned_count=0, message=( f"{assigned_count} Magento-assigned SKU(s) have no gallery images. " "Import images on Catalog Import first." ), ) queue_repo.mark_done(queue_id) logger.info( "[queue=%s] stage=done: images_only — %d assigned, 0 with gallery images", queue_id, assigned_count, ) session.commit() return True else: logger.info("[queue=%s] stage=snapshots: building from master catalog", queue_id) from magento.master_catalog_rows import build_magento_snapshot_rows_from_master snapshot_rows_preflight = build_magento_snapshot_rows_from_master( session, skus=sku_list, only_assigned=False, connection_id=connection_id, expand_relations=expand_relations, ) import pandas as pd snapshot_df = pd.DataFrame(snapshot_rows_preflight) logger.info("[queue=%s] stage=snapshots: %d master rows built", queue_id, len(snapshot_df)) if images_only: sku_list = [str(r.get("sku", "")).strip() for r in snapshot_rows if str(r.get("sku", "")).strip()] else: # Debug: trace specific SKUs through snapshot stage (before normalize) if _DEBUG or effective_trace_skus: for idx, row in snapshot_df.iterrows(): sku = str(row.get("sku", "")).strip() if not _trace_sku(sku, effective_trace_skus) and not _DEBUG: continue if _trace_sku(sku, effective_trace_skus): logger.info( "[queue=%s] trace=snapshot sku=%s product_type=%r variant_list=%r configurable_variations=%r configurable_variation_labels=%r", queue_id, sku, row.get("product_type"), (str(row.get("variant_list") or ""))[:100], (str(row.get("configurable_variations") or ""))[:100], (str(row.get("configurable_variation_labels") or ""))[:80], ) if _DEBUG: configurable_count = sum(1 for _, r in snapshot_df.iterrows() if "configurable" in str(r.get("product_type", "")).lower()) with_variant_list = sum(1 for _, r in snapshot_df.iterrows() if str(r.get("variant_list") or "").strip()) logger.info("[queue=%s] debug=snapshots configurable_count=%d with_variant_list=%d total=%d", queue_id, configurable_count, with_variant_list, len(snapshot_df)) logger.info("[queue=%s] stage=normalize: running", queue_id) normalize_started = time.perf_counter() cfg = load_magento_config() registry_codes = load_attribute_registry_codes(load_db_config().attribute_registry_path) registry_options = load_attribute_registry_options(load_db_config().attribute_registry_path) from db.channel_remote_attributes import ( expand_magento_allowed_attribute_codes, sync_option_values_from_master, ) registry_codes = expand_magento_allowed_attribute_codes(session, registry_codes) if not dry_run and sync_options_from_master_enabled: try: logger.info("[queue=%s] stage=normalize: sync_options_from_master start", queue_id) opt_sync = sync_option_values_from_master( session, channel_type="magento", connection_id=connection_id, dry_run=False, attribute_codes=selected_attribute_codes, ) session.commit() logger.info( "[queue=%s] stage=normalize: sync_options_from_master done aliases=%s elapsed_ms=%s", queue_id, opt_sync.get("aliases"), opt_sync.get("elapsed_ms"), ) except Exception as opt_exc: logger.warning("[queue=%s] sync_options_from_master failed: %s", queue_id, opt_exc) elif not dry_run and not sync_options_from_master_enabled: logger.info("[queue=%s] sync_options_from_master: skipped by queue option", queue_id) from magento.catalog_normalize import normalize_master_catalog_df logger.info( "[queue=%s] stage=normalize: dataframe start rows=%d", queue_id, len(snapshot_df), ) df_started = time.perf_counter() normalized_df, quarantine_df = normalize_master_catalog_df( snapshot_df, cfg, allowed_attribute_codes=registry_codes or None, allowed_attribute_options=registry_options or None, ) logger.info( "[queue=%s] stage=normalize: dataframe done rows=%d elapsed_ms=%d", queue_id, len(normalized_df), int((time.perf_counter() - df_started) * 1000), ) if len(quarantine_df) and len(normalized_df) < len(snapshot_df): reasons: Dict[str, int] = {} for _, qrow in quarantine_df.iterrows(): reason = str(qrow.get("reason") or "unknown") reasons[reason] = reasons.get(reason, 0) + 1 logger.warning( "[queue=%s] normalize quarantined %d/%d row(s): %s", queue_id, len(quarantine_df), len(snapshot_df), reasons, ) # Do NOT shorten image URLs for Magento sync: magento_media_map uses # the original gallery URL as remote_url for matching. snapshot_rows = normalized_df.to_dict(orient="records") logger.info( "[queue=%s] stage=normalize: done, %d rows ready elapsed_ms=%d", queue_id, len(snapshot_rows), int((time.perf_counter() - normalize_started) * 1000), ) if not dry_run: from db.channel_attribute_provision import provision_magento_configurable_catalog prov_started = time.perf_counter() logger.info("[queue=%s] stage=provision_configurable: start", queue_id) cfg_prov = provision_magento_configurable_catalog( session, connection_id, snapshot_rows=snapshot_rows, dry_run=False, ) logger.info( "[queue=%s] stage=provision_configurable: done elapsed_ms=%d created_options=%s created_attributes=%s errors=%s", queue_id, int((time.perf_counter() - prov_started) * 1000), cfg_prov.get("created_options"), cfg_prov.get("created_attributes"), len(cfg_prov.get("errors") or []), ) if cfg_prov.get("created_options") or cfg_prov.get("created_attributes") or cfg_prov.get("errors"): logger.info("[queue=%s] configurable_catalog_provision: %s", queue_id, cfg_prov) # Debug: trace specific SKUs after normalize if _DEBUG or effective_trace_skus: for row in snapshot_rows: sku = str(row.get("sku", "")).strip() if not _trace_sku(sku, effective_trace_skus) and not _DEBUG: continue if _trace_sku(sku, effective_trace_skus): logger.info( "[queue=%s] trace=normalized sku=%s product_type=%r variant_list=%r configurable_variations=%r configurable_attributes=%r", queue_id, sku, row.get("product_type"), (str(row.get("variant_list") or ""))[:100], (str(row.get("configurable_variations") or ""))[:100], str(row.get("configurable_attributes") or ""), ) if _DEBUG: configurable_parent_count = sum( 1 for r in snapshot_rows if "configurable" in str(r.get("product_type", "")).lower() and (str(r.get("variant_list") or "").strip() or str(r.get("configurable_variations") or "").strip()) ) logger.info("[queue=%s] debug=normalized configurable_parent_count=%d total_rows=%d", queue_id, configurable_parent_count, len(snapshot_rows)) # Hold SKUs that are under pending deletion review snapshot_rows = _filter_pending_deletion_review( session, connection_id, snapshot_rows, queue_id ) if limit_skus: allowed = {str(s).strip() for s in limit_skus if str(s).strip()} before = len(snapshot_rows) snapshot_rows = [ row for row in snapshot_rows if str(row.get("sku", "")).strip() in allowed ] if before != len(snapshot_rows): logger.info( "[queue=%s] clipped snapshot rows to explicit limit_skus: %d -> %d", queue_id, before, len(snapshot_rows), ) # Resumable sync: reuse job if queue was reset (crash recovery) existing_job_id = item.get("job_id") if existing_job_id: job_id = existing_job_id logger.info("[queue=%s] resuming job_id=%s (pending items will be executed)", queue_id, job_id) else: job_id = sync_repo.create_sync_job( MagentoSyncJobCreate( connection_id=connection_id, dry_run=dry_run, total_count=len(snapshot_rows), ) ) queue_repo.set_job_id(queue_id, job_id) session.commit() sync_repo.update_sync_job(job_id, status="running", started_at=datetime.utcnow()) session.commit() # release job row lock before long planning transaction def _report_progress(completed: int, total: int, success: int, errors: int) -> None: # Commit sync work (pending plan / item outcomes) without touching # magento_sync_jobs counters on this session. if images_only: from app.jobs.channel_jobs import report_channel_image_push_outcome report_channel_image_push_outcome( session, queue_id, planned_count=total, success_count=success, error_count=errors, message=( f"Image push in progress: {completed}/{total} actions " f"({success} ok, {errors} failed)" ), ) session.commit() try: _persist_magento_sync_job_progress( job_id, total_count=total, success_count=success, error_count=errors, ) except Exception: # Progress is best-effort; never abort a successful sync for dashboard counters. logger.warning( "[queue=%s] progress update failed for job_id=%s (non-fatal)", queue_id, job_id, exc_info=True, ) oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) try: BASE_IMAGE_URL = __import__("main", fromlist=["BASE_IMAGE_URL"]).BASE_IMAGE_URL except (ImportError, AttributeError): BASE_IMAGE_URL = "" def resolve_url(url): if not url or not BASE_IMAGE_URL: return url from db.image_urls import normalize_external_image_url url = normalize_external_image_url(url) if url.startswith(BASE_IMAGE_URL) and "/api/i/" in url: slug = url.rstrip("/").split("/")[-1] rec = resolve_slug(ingest_repo, slug) return rec.destination_url if rec else url return url sync_svc = MagentoSyncService( api, conn, sync_state_repo=state_repo, media_map_repo=media_repo, sync_job_repo=sync_repo, sync_pending_repo=pending_repo, override_repo=override_repo, version_repo=version_repo, notification_repo=notification_repo, attr_cache_repo=attr_registry_repo, category_cache_repo=cat_registry_repo, store_registry_repo=store_registry_repo, baseline_repo=baseline_repo, attr_set_attrs_repo=attr_set_attrs_repo, attr_set_registry_repo=attr_set_registry_repo, resolve_image_url=resolve_url, source_codes_provider=_source_codes_provider, ) catalog_media = catalog_repo.get_media_hashes_for_skus(connection_id, sku_list) catalog_state = catalog_repo.get_catalog_state_for_skus(connection_id, sku_list) execute_skus = None if limit_skus: execute_skus = { str(s).strip() for s in limit_skus if str(s).strip() } if force_upsert and not force_upsert_skus: force_upsert_skus = [ str(r.get("sku") or "").strip() for r in snapshot_rows if str(r.get("sku") or "").strip() ] logger.info( "[queue=%s] stage=sync: starting incremental sync job_id=%s dry_run=%s force_upsert=%s", queue_id, job_id, dry_run, bool(force_upsert_skus), ) def _abort_if_superseded() -> bool: if not queued_at or allow_parallel_with_newer_queue_items: return False return queue_repo.has_newer_active_item(queue_id, connection_id, queued_at) result = sync_svc.sync_incremental( connection_id, snapshot_rows, job_id=job_id, dry_run=dry_run, state_repo=state_repo, job_repo=sync_repo, override_repo=override_repo, version_repo=version_repo, notification_repo=notification_repo, notify_email=options.get("notify_email"), catalog_state_media_by_sku=catalog_media, catalog_state_by_sku=catalog_state, force_relations=force_relations, replace_options=replace_options, force_full_sync=force_full_sync, debug_output_path=debug_output_path, trace_skus=list(effective_trace_skus) if effective_trace_skus else None, force_upsert_skus=list(force_upsert_skus) if force_upsert_skus else None, images_only=images_only, products_only=products_only, force_images=force_images, execute_skus=execute_skus, selected_attribute_codes=set(selected_attribute_codes or []), use_connection_pending_table=use_connection_pending_table, progress_callback=_report_progress, abort_callback=_abort_if_superseded, ) if result.get("status") == "cancelled": message = result.get("message") or "Cancelled because a newer Magento queue item superseded this run" queue_repo.mark_cancelled(queue_id, message) sync_repo.update_sync_job(job_id, status="cancelled", notes=message, finished_at=datetime.utcnow()) session.commit() logger.warning("[queue=%s] stage=cancelled: %s", queue_id, message) return True if images_only: from app.jobs.channel_jobs import report_channel_image_push_outcome planned = int(result.get("planned_count") or 0) ok = int(result.get("success_count") or 0) err = int(result.get("error_count") or 0) with_images = len(snapshot_rows) assigned = len(sku_list) if planned == 0 and with_images > 0: msg = ( f"{with_images} SKU(s) with images checked; no uploads needed " "(already in sync or product missing in Magento — link SKU / run full push first)." ) elif planned == 0: msg = "No image uploads were planned for this run." elif err: msg = f"Image push finished: {ok} succeeded, {err} failed (of {planned} planned)." else: msg = f"Image push finished: {ok} of {planned} image action(s) succeeded." report_channel_image_push_outcome( session, queue_id, assigned_sku_count=assigned, skus_with_images=with_images, planned_count=planned, success_count=ok, error_count=err, message=msg, extra={"planned_count": planned}, ) sync_repo.update_sync_job(job_id, notes=msg) queue_repo.mark_done(queue_id) # Commit before pull_after so waiters/pollers see terminal status promptly. # pull_after is verification only and can take longer than wait timeouts. session.commit() logger.info("[queue=%s] stage=done: job_id=%s success=%s errors=%s", queue_id, job_id, result.get("success_count"), result.get("error_count")) run_pull_after = options.get("run_pull_after", True) if run_pull_after and not dry_run and not images_only: logger.info("[queue=%s] stage=pull_after: pulling to verify and update catalog_state", queue_id) from app.jobs.magento_pull import run_pull run_pull(connection_id, date.today()) # since=today captures newly created products logger.info("[queue=%s] stage=pull_after: done", queue_id) session.commit() except Exception as e: logger.exception("[queue=%s] stage=FAILED: %s", queue_id, e) # Deadlock/flush failures leave the session in PendingRollbackError; # always rollback before writing the failed queue row. try: session.rollback() except Exception: pass try: queue_repo.mark_failed(queue_id, str(e)) session.commit() except Exception: logger.exception("[queue=%s] failed to mark queue item failed", queue_id) try: session.rollback() except Exception: pass return True return True def main() -> int: parser = argparse.ArgumentParser(description="Process magento_sync_queue") parser.add_argument("--once", action="store_true", help="Process one item and exit") args = parser.parse_args() if args.once: if run_one(): return 0 return 0 while run_one(): pass return 0 if __name__ == "__main__": sys.exit(main())