How It Works

1. Integration Layer: Provider-specific integrations like Unicommerce, Shopify, and other future providers connect to Ecommerce Core.

2. Core Layer: The middleware that handles:

• Data Mapping: Links ERPNext items to platform items
• Synchronization: Manages data flow between systems
• Controllers: Provides reusable business logic
• Logging: Tracks all operations and enables retries

3. ERPNext Layer: Your ERPNext instance where the actual business data lives.

Core Components

1. Data Mapping Layer

The mapping layer maintains referential integrity between ERPNext entities and their external platform counterparts through a sophisticated linking table architecture.

Ecommerce Item Doctype

The Ecommerce Item doctype serves as the primary mapping table, establishing bidirectional relationships between ERPNext items and platform entities:

Technical Implementation:

• Indexed on erpnext_item_code, integration, and integration_item_code for optimized lookup performance
• Supports variant item hierarchies with parent-child relationships
• Tracks synchronization timestamps to enable incremental updates
• Enforces referential integrity through database constraints

2. Synchronization Framework

The synchronization framework provides robust, fault-tolerant data exchange with comprehensive error handling and retry capabilities.

Ecommerce Integration Log

The Ecommerce Integration Log doctype implements a centralized logging mechanism for all integration operations:

Features:

• Automatic log retention management (120-day default)
• Retry mechanism integration with UI controls
• Request/response debugging capabilities
• Status tracking for scheduled operations
• Multi-provider log segregation

3. Base Controllers & Abstractions

The framework provides abstract base classes that define integration contracts while encapsulating common functionality.

SettingController

Abstract base class for provider-specific settings doctypes:

class SettingController(Document):
    def is_enabled(self) -> bool:
        """Check if integration is enabled or not."""
        raise NotImplementedError()
    
    def get_erpnext_warehouses(self) -> list[ERPNextWarehouse]:
        """Get configured ERPNext warehouses for integration."""
        raise NotImplementedError()
    
    def get_erpnext_to_integration_wh_mapping(self) -> dict[ERPNextWarehouse, IntegrationWarehouse]:
        """Map ERPNext warehouses to platform warehouse identifiers."""
        raise NotImplementedError()
    
    def get_integration_to_erpnext_wh_mapping(self) -> dict[IntegrationWarehouse, ERPNextWarehouse]:
        """Reverse mapping for platform warehouse identifiers."""
        raise NotImplementedError()

Provider Implementation Pattern:

1. Extend SettingController in provider-specific settings doctype
2. Implement warehouse mapping configuration
3. Enable/disable integration functionality
4. Configure synchronization intervals

EcommerceCustomer

Comprehensive customer synchronization controller:

class EcommerceCustomer:
    def __init__(self, customer_id: str, customer_id_field: str, integration: str):
        """Initialize with platform customer ID and target integration."""
    
    def is_synced(self) -> bool:
        """Check if customer exists in ERPNext."""
    
    def get_customer_doc(self):
        """Retrieve ERPNext customer document."""
    
    def sync_customer(self, customer_name: str, customer_group: str) -> None:
        """Create customer in ERPNext if not exists."""
    
    def create_customer_address(self, address: dict[str, str]) -> None:
        """Create address linked to customer."""
    
    def create_customer_contact(self, contact: dict[str, str]) -> None:
        """Create contact linked to customer."""

Synchronization Workflow:

1. Check for existing customer using is_synced()
2. Create new customer if not exist using sync_customer()
3. Synchronize address information with create_customer_address()
4. Synchronize contact details with create_customer_contact()

4. Inventory Management

The inventory management subsystem provides real-time stock synchronization with support for warehouse hierarchies and delta updates.

Core Inventory Functions

get_inventory_levels(warehouses: tuple[str], integration: str)

Retrieves items requiring inventory updates based on bin modification timestamps:

def get_inventory_levels(warehouses: tuple[str], integration: str) -> list[_dict]:
    """
    Returns items where Bin.modified > EcommerceItem.inventory_synced_on
    for specified warehouses and integration.
    
    Returns: list of _dict containing:
        - ecom_item: Ecommerce Item name
        - item_code: ERPNext item code
        - integration_item_code: Platform item identifier
        - variant_id: Platform variant identifier
        - actual_qty: Current stock quantity
        - warehouse: Warehouse location
        - reserved_qty: Reserved quantity
    """

get_inventory_levels_of_group_warehouse(warehouse: str, integration: str)

Consolidates inventory from child warehouses for group warehouse mappings:

def get_inventory_levels_of_group_warehouse(warehouse: str, integration: str):
    """
    Aggregates inventory from warehouse hierarchy.
    Consolidates all child warehouse quantities for parent warehouse.
    """

update_inventory_sync_status(ecommerce_item, time=None)

Updates synchronization timestamp to prevent duplicate processing:

def update_inventory_sync_status(ecommerce_item, time=None):
    """Marks inventory as synchronized to specified or current time."""

Performance Optimization:

• Query builder implementation for efficient database operations
• Incremental updates based on modification timestamps
• Support for high-volume inventory sync operations
• Warehouse hierarchy traversal for group warehouses

5. Scheduling Engine

The scheduling framework enables configurable, interval-based synchronization tasks with automatic timestamp management.

need_to_run(setting, interval_field, timestamp_field)

Implements configurable scheduled event gating:

def need_to_run(setting, interval_field, timestamp_field) -> bool:
    """
    Determines if scheduled task should execute based on:
    - Configured interval (in minutes)
    - Last execution timestamp
    - Current time
    
    Returns True if task should run and updates timestamp to now().
    Returns False if interval has not elapsed.
    
    Assumptions:
        - interval_field stores interval in minutes
        - timestamp_field is DateTime field
        - Called from scheduled job with frequency < lowest interval
    """

Usage Pattern:

# In scheduled job
if need_to_run("Unicommerce Settings", "inventory_sync_frequency", "last_inventory_sync"):
    sync_inventory()

Scheduling Configuration:

• Provider-specific settings define intervals in minutes
• Automatic timestamp management prevents duplicate execution
• Supports multiple concurrent scheduled tasks
• Flexible interval configuration per integration