1C Integration with Magento 2

1C Integration with Magento 2

Bidirectional Magento 2 and 1C synchronization is standard task for Russian e-commerce: catalog and prices from 1C, orders to 1C, real-time stock. More complex than OpenCart due to Magento-specific concepts (attribute sets, configurable products, MSI).

Integration Architecture

1C ←→ Integration Layer ←→ Magento 2

Integration Layer:
- Queue (RabbitMQ / Redis)
- Transformer (1C format ↔ Magento format)
- Error Handler
- Retry Logic

Direct 1C → Magento REST API connection works but is unreliable under load. Intermediate layer with queue is right approach.

1C HTTP Service for Data Export

// 1C: HTTP Service ProductsService
// GET /products Method
Function GetProducts(Request)
    ProductArray = New Array;

    Selection = Catalogs.Nomenclature.Select();
    While Selection.Next() Do
        If NOT Selection.IsFolder Then
            Product = New Structure;
            Product.Insert("sku",      String(Selection.Ref.UniqueId()));
            Product.Insert("name",     Selection.Name);
            Product.Insert("price",    InformationRegisters.NomenclaturePrices.GetLast(Selection.Ref).Price);
            Product.Insert("quantity", AccumulationRegisters.GoodsInWarehouses.ProductBalance(Selection.Ref));
            ProductArray.Add(Product);
        EndIf;
    EndDo;

    Response = New HTTPServiceResponse(200);
    Response.SetBodyFromString(Format.JSON.Write(ProductArray));
    Return Response;
EndFunction

Import from 1C to Magento 2 (Python integration worker)

import requests
import json
from magento.models.product import Product

class OneCMagentoSync:
    def __init__(self, m2_url, m2_token, onec_url, onec_auth):
        self.m2_url = m2_url
        self.m2_headers = {"Authorization": f"Bearer {m2_token}", "Content-Type": "application/json"}
        self.onec_url = onec_url
        self.onec_auth = onec_auth

    def sync_products(self):
        # Get products from 1C
        response = requests.get(f"{self.onec_url}/products", auth=self.onec_auth)
        products_1c = response.json()

        for product_data in products_1c:
            self.upsert_product(product_data)

    def upsert_product(self, data: dict):
        sku = data['sku']

        # Check existence in Magento
        check = requests.get(f"{self.m2_url}/rest/V1/products/{sku}", headers=self.m2_headers)

        payload = {
            "product": {
                "sku": sku,
                "name": data['name'],
                "price": float(data['price']),
                "status": 1,
                "type_id": "simple",
                "attribute_set_id": 4,
                "extension_attributes": {
                    "stock_item": {
                        "qty": int(data['quantity']),
                        "is_in_stock": int(data['quantity']) > 0
                    }
                }
            }
        }

        if check.status_code == 200:
            # Update
            requests.put(f"{self.m2_url}/rest/V1/products/{sku}",
                        headers=self.m2_headers, json=payload)
        else:
            # Create
            requests.post(f"{self.m2_url}/rest/V1/products",
                         headers=self.m2_headers, json=payload)

Stock Synchronization (MSI)

Magento 2.3+ uses Multi-Source Inventory. Update stock via MSI API:

def update_stock_msi(self, sku: str, source_code: str, qty: float):
    payload = {
        "sourceItems": [{
            "sku": sku,
            "source_code": source_code,
            "quantity": qty,
            "status": 1 if qty > 0 else 0
        }]
    }
    requests.post(f"{self.m2_url}/rest/V1/inventory/source-items",
                 headers=self.m2_headers, json=payload)

Exporting Orders to 1C

def export_orders_to_1c(self):
    # Get unprocessed orders
    response = requests.get(
        f"{self.m2_url}/rest/V1/orders",
        params={
            "searchCriteria[filter_groups][0][filters][0][field]": "status",
            "searchCriteria[filter_groups][0][filters][0][value]": "pending",
            "searchCriteria[pageSize]": 50
        },
        headers=self.m2_headers
    )

    orders = response.json()['items']

    for order in orders:
        order_1c = self.transform_order(order)

        # Send to 1C
        result = requests.post(
            f"{self.onec_url}/orders",
            json=order_1c,
            auth=self.onec_auth,
            timeout=10
        )

        if result.status_code == 200:
            # Mark order as transmitted to 1C
            requests.post(
                f"{self.m2_url}/rest/V1/orders/{order['entity_id']}/comments",
                headers=self.m2_headers,
                json={"statusHistory": {"comment": "Transmitted to 1C", "status": "processing"}}
            )

Queue for Reliability

import pika  # RabbitMQ

def publish_to_queue(self, event_type: str, data: dict):
    connection = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URL))
    channel = connection.channel()
    channel.queue_declare(queue='magento_1c_sync', durable=True)
    channel.basic_publish(
        exchange='',
        routing_key='magento_1c_sync',
        body=json.dumps({"type": event_type, "data": data}),
        properties=pika.BasicProperties(delivery_mode=2)  # persistent
    )
    connection.close()

Timelines

Basic synchronization (catalog + stock in one direction) — 10–14 days. Bidirectional integration with orders, queue, and monitoring — 3–4 weeks.