Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/VisualGraphxLLC/API-HUB/llms.txt

Use this file to discover all available pages before exploring further.

The adapter framework is the layer that decouples the import orchestrator from supplier-specific protocol details. Every supplier, regardless of whether it speaks SOAP, REST, or GraphQL, presents the same four abstract methods to the orchestrator. The registry maps a plain string stored in the database to the correct class at runtime, so adding a new supplier type never requires changes to the import pipeline.

​
BaseAdapter

BaseAdapter is an abstract base class defined in backend/modules/import_jobs/base.py. Every adapter must subclass it and implement three abstract methods. A fourth method, discover_closeouts, has a default implementation that raises NotImplementedError and only needs to be overridden by adapters whose suppliers support closeout feeds. 
class BaseAdapter(ABC):
    product_type: str  # subclasses set "apparel" or "print"

    def __init__(self, supplier: Supplier, db: AsyncSession):
        self.supplier = supplier
        self.db = db

    @abstractmethod
    async def discover(
        self,
        mode: DiscoveryMode,
        limit: Optional[int] = None,
        explicit_list: Optional[List[str]] = None,
    ) -> List[ProductRef]:
        """Return product references for the given discovery mode."""

    @abstractmethod
    async def hydrate_product(self, ref: ProductRef) -> ProductIngest:
        """Fetch full product details and normalize to ProductIngest."""

    @abstractmethod
    async def discover_changed(self, since: datetime) -> List[ProductRef]:
        """Return references for products changed since the given timestamp."""

    async def discover_closeouts(self) -> List[ProductRef]:
        raise NotImplementedError
The orchestrator always calls discover() first to get a list of ProductRef objects, then calls hydrate_product() for each ref to fetch full details. ProductRef is a lightweight model holding just the supplier_sku and an optional part_id.

​
product_type

Set product_type as a class-level attribute to either "apparel" or "print". The normalizer uses this to route the ProductIngest through the correct schema branch. PromoStandards adapters default to "apparel"; FourOverAdapter and OPSAdapter use "print".

​
DiscoveryMode

DiscoveryMode is a string enum. The import orchestrator accepts a mode string from the API request and passes the resolved enum value to adapter.discover(). 
class DiscoveryMode(str, Enum):
    FULL             = "full"
    DELTA            = "delta"
    FIRST_N          = "first_n"
    EXPLICIT_LIST    = "explicit_list"
    FILTERED_SAMPLE  = "filtered_sample"
    FULL_SELLABLE    = "full_sellable"
    CLOSEOUTS        = "closeouts"
Discovers every product the supplier exposes. This is the most data-intensive mode and is typically run once at onboarding or after a prolonged gap. On PromoStandards adapters, FULL is not explicitly implemented — use FULL_SELLABLE instead, which calls getProductSellable.
Discovers products modified since supplier.last_delta_sync (or last_full_sync if last_delta_sync is null, or 2000-01-01 if neither is set). Calls discover_changed() internally. Use this for daily or hourly incremental syncs.On suppliers that do not expose a modified-since endpoint (e.g. 4Over), DELTA falls back to full discovery automatically.
Returns the first limit products from the supplier’s sellable list. The orchestrator passes limit from the API request body. Minimum is 1, maximum is 10,000.Use this mode for a smoke-test import when adding a new supplier: mode=first_n, limit=5.
Imports only the product IDs you list in explicit_list. Requires the explicit_list field in the request body; the request is rejected at the schema layer if the list is absent.
{
  "mode": "explicit_list",
  "explicit_list": ["PC61", "PC54", "K500"]
}
Fetches the full sellable list from the supplier, then filters it to the IDs in explicit_list (or protocol_config.explicit_list if not provided in the request). Useful when you want to verify that specific SKUs are in the sellable feed before importing them.
Calls getProductSellable (for PS adapters) and returns every product flagged as sellable. This is the standard mode for full PromoStandards syncs. Products with isSellable: false are excluded.
Calls discover_closeouts(), which maps to getProductCloseOut on PS adapters. Only implement this if the supplier’s API supports a closeout endpoint. The base implementation raises NotImplementedError.

​
Error types

Three error types control how the import pipeline handles failures. All inherit from AdapterError, which accepts an optional code string alongside the message.

AuthError

Authentication failed. This is a fatal error for the import job. The job stops immediately and marks the supplier as needing attention. Raised when PS SOAP fault codes 100, 104, or 110 are received, or when required credentials are missing from auth_config.

SupplierError

The supplier returned an error for a specific product. The pipeline logs the error and skips that product, continuing with the rest of the batch. One broken product does not abort a sync of thousands.

TransientError

A network timeout, connection reset, or 5xx response that may succeed on retry. The pipeline retries up to three times with exponential backoff: delays of 4 s, 2 s, and 1 s (2 ** (2 - attempt) seconds). If all retries fail, the product is skipped and logged.

​
Retry logic

# Pseudocode — actual implementation in the import orchestrator
for attempt in range(3):
    try:
        ingest = await adapter.hydrate_product(ref)
        break
    except TransientError:
        delay = 2 ** (2 - attempt)   # 4s, 2s, 1s
        await asyncio.sleep(delay)
    except SupplierError:
        log_and_skip(ref)
        break
    except AuthError:
        abort_job()
        raise

​
The adapter registry

The registry is a module-level dict in backend/modules/import_jobs/registry.py. It maps the string stored in suppliers.adapter_class to the concrete class. 
ADAPTERS: dict[str, Type[BaseAdapter]] = {}

def register_adapter(name: str, cls: Type[BaseAdapter]) -> None:
    if not issubclass(cls, BaseAdapter):
        raise TypeError(f"{cls!r} is not a BaseAdapter subclass")
    ADAPTERS[name] = cls

​
Self-registration pattern

Each adapter module calls register_adapter at the bottom of its file, immediately after the class definition. This means the adapter registers itself the moment the module is imported — no central configuration file needed. 
# At the bottom of promostandards/adapter.py
register_adapter("PromoStandardsAdapter", PromoStandardsAdapter)

# At the bottom of promostandards/sanmar_adapter.py
register_adapter("SanMarAdapter", SanMarAdapter)

# At the bottom of rest_connector/fourover_adapter.py
register_adapter("FourOverAdapter", FourOverAdapter)
main.py imports all adapter modules on startup. By the time FastAPI mounts the import routes, ADAPTERS is fully populated.

​
Registry lookup

def get_adapter(supplier, db: AsyncSession) -> BaseAdapter:
    adapter_key = getattr(supplier, "adapter_class", None)
    if not adapter_key:
        raise AdapterNotConfiguredError(
            f"Supplier {supplier.name!r} has no adapter_class set"
        )
    cls = ADAPTERS.get(adapter_key)
    if cls is None:
        raise AdapterNotRegisteredError(
            f"adapter_class {adapter_key!r} not registered. "
            f"Known: {sorted(ADAPTERS)}"
        )
    return cls(supplier=supplier, db=db)
Two distinct errors are raised so operators can tell the difference between a missing database value (AdapterNotConfiguredError) and a value that references a class that was never registered (AdapterNotRegisteredError). The latter error message includes the list of known adapter names to help diagnose typos.

​
Choosing a discovery mode

ScenarioRecommended mode
First import after adding a supplierfirst_n with limit=5
Full catalog load at onboardingfull_sellable
Daily incremental syncdelta
Re-import specific SKUs after a data issueexplicit_list
Verify specific SKUs are in the sellable feedfiltered_sample
Import clearance productscloseouts
Always run first_n with a small limit before scheduling full or delta syncs. It validates credentials against the live SOAP or REST endpoint and surfaces any authentication issues before they affect a large batch.

​
Triggering an import via API

POST /api/suppliers/{supplier_id}/import
Content-Type: application/json

{
  "mode": "first_n",
  "limit": 5
}
The endpoint returns 202 Accepted immediately. The import runs in a background task managed by FastAPI’s BackgroundTasks. The response includes a sync_job_id you can use to poll status: 
GET /api/suppliers/{supplier_id}/sync-jobs
If a job with the same supplier_id and mode is already pending or running, the endpoint returns 409 Conflict instead of queuing a second job. This prevents duplicate concurrent imports for the same supplier and mode combination.

Build docs developers (and LLMs) love

Get started for freeTalk to us