How I Automated CJ Dropshipping Order Fulfillment with a Python Script

The 2 AM manual fulfillment trap

When we launched Nyxeris—our luxury workspace hardware and EDC project—the initial surge of customer orders was thrilling. A customer in Austin purchased our CNC aluminum wireless charging dock; twenty minutes later, a designer in Frankfurt ordered an anodized monitor riser and desk mat.

The euphoria lasted roughly seventy-two hours. Then fulfillment reality hit.

Because we architected Nyxeris as a self-hosted platform—a high-performance Next.js storefront backed by an embedded SQLite database running on a lightweight Linux VPS—we intentionally avoided closed SaaS ecosystems like Shopify or BigCommerce. Our manufacturing and fulfillment partner of choice was CJ Dropshipping, chosen for their vetted international warehouses, custom packaging options, and fast shipping lines like CJ Packet and USPS Priority.

However, CJ Dropshipping's official plug-and-play integrations cater almost exclusively to standard Shopify or WooCommerce stores. For an independent headless stack, our fulfillment workflow looked like an agonizing ritual every night at 2:00 AM:

1. Open our SQLite admin console, query the day's paid transactions, and copy customer names, addresses, and postal codes into the clipboard.
2. Log into the CJ Dropshipping web portal, navigate past marketing popups, and open DropShipping Center → Imported Orders → Add Manual Order.
3. Paste customer address line 1, city, state, postal code, and phone number into CJ's multi-step form.
4. Search our supplier catalog, locate the master product listing, open the variant picker, and hunt for the matching configuration (for example, "Matte Obsidian / 65W GaN / US Plug").
5. Submit the manual order, confirm the shipping method, copy CJ's internal order number, and paste it back into our local database.
6. Log back in two days later to manually scrape the tracking number and paste it into a customer dispatch email.

At five orders a day, this is a tedious nuisance. At thirty orders a day, it is an operational nightmare. Transpose two numbers in a German postal code or omit an apartment number, and a $75 custom parcel gets rejected at customs and routed back across an ocean. We were spending nearly two hours every day acting as human copy-paste middleware.

The solution was obvious: eliminate the browser interface entirely and write an autonomous Python daemon that talks directly to CJ Dropshipping's Open API 2.0.

Authenticating with CJ Dropshipping Open API 2.0

CJ Dropshipping exposes a dedicated REST interface called Open API 2.0 (located at https://developers.cjdropshipping.com/api2.0/v1). Unlike complex OAuth2 authorization code flows that require interactive browser handshakes and redirect callbacks, CJ's authentication protocol is server-friendly: you authenticate using a private API Key generated inside your merchant dashboard.

To establish an authorized session, your backend sends an HTTP POST request to /authentication/getAccessToken containing your API key in the JSON payload:

{
  "apiKey": "CJ_API_KEY_SECRET_STRING"
}

Upon validation, CJ responds with an authorization payload containing an accessToken (a secure JWT token) and an accessTokenExpiryDate timestamp (typically valid for 15 to 30 days):

{
  "code": 200,
  "result": true,
  "message": "Success",
  "data": {
    "accessToken": "eyJhbGciOiJIUzI1NiIsIn...",
    "accessTokenExpiryDate": "2026-10-15 11:24:00"
  }
}

Every subsequent request to CJ—whether querying products, submitting orders, or fetching logistics updates—must provide this access token inside the custom HTTP header CJ-Access-Token.

Because requesting a fresh token on every order consumes unnecessary network round-trips and risks hitting CJ's rate limits, our Python script manages an on-disk token cache with automatic proactive refreshes:

import json
import logging
import os
import time
from datetime import datetime
import httpx

logger = logging.getLogger("fulfillment.auth")

class CJAuthManager:
    API_BASE = "https://developers.cjdropshipping.com/api2.0/v1"
    CACHE_FILE = "data/cj_token_cache.json"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self._token = None
        self._expiry_timestamp = 0
        self._load_cached_token()

    def _load_cached_token(self):
        if os.path.exists(self.CACHE_FILE):
            try:
                with open(self.CACHE_FILE, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    self._token = data.get("access_token")
                    self._expiry_timestamp = data.get("expires_at", 0)
            except Exception as err:
                logger.warning(f"Could not read local token cache: {err}")

    def _save_cached_token(self, token: str, expires_at: float):
        self._token = token
        self._expiry_timestamp = expires_at
        os.makedirs(os.path.dirname(self.CACHE_FILE), exist_ok=True)
        with open(self.CACHE_FILE, "w", encoding="utf-8") as f:
            json.dump({"access_token": token, "expires_at": expires_at}, f, indent=2)

    def get_valid_token(self) -> str:
        if self._token and time.time() < (self._expiry_timestamp - 43200):
            return self._token

        logger.info("Requesting new access token from CJ Dropshipping API...")
        resp = httpx.post(
            f"{self.API_BASE}/authentication/getAccessToken",
            json={"apiKey": self.api_key},
            timeout=15.0
        )
        resp.raise_for_status()
        body = resp.json()

        if not body.get("result"):
            raise RuntimeError(f"CJ authentication failed: {body.get('message')}")

        token_data = body["data"]
        new_token = token_data["accessToken"]
        expiry_str = token_data.get("accessTokenExpiryDate")
        
        expiry_dt = datetime.strptime(expiry_str, "%Y-%m-%d %H:%M:%S")
        self._save_cached_token(new_token, expiry_dt.timestamp())
        logger.info("CJ access token refreshed and cached successfully.")
        return new_token

SQLite schema and the SKU translation layer

The most critical conceptual hurdle in automated dropshipping is the SKU mapping disconnect.

In our store's public catalog, we assign clean, human-readable SKUs tailored for branding and analytics, such as NYX-DOCK-BLK (Nyxeris Induction Headphone Dock, Obsidian). However, CJ Dropshipping's warehouse management system does not know what NYX-DOCK-BLK means. CJ's inventory database identifies items using two specific identifiers:

• PID (Product ID): Represents the parent catalog item (e.g., CJ-PID-849201).
• VID (Variant ID): Represents the exact SKU variant—size, color, specifications, and regional warehouse origin (e.g., CJ-VAR-994102-BLK).

If you submit an order payload containing your internal SKU, CJ's API rejects the entire payload with error code 1002 ("Variant ID does not exist"). The automated fulfillment engine requires an explicit translation layer that maps internal storefront SKUs to supplier variant IDs.

Here is our SQLite relational schema supporting the store orders, individual line items, and the SKU translation bridge:

-- Store customer orders
CREATE TABLE orders (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    order_number TEXT NOT NULL UNIQUE,
    customer_name TEXT NOT NULL,
    email TEXT NOT NULL,
    phone TEXT NOT NULL,
    address_line1 TEXT NOT NULL,
    address_line2 TEXT,
    city TEXT NOT NULL,
    state TEXT NOT NULL,
    postal_code TEXT NOT NULL,
    country_code TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'PAID',
    cj_order_id TEXT,
    tracking_number TEXT,
    carrier TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    fulfilled_at DATETIME
);

-- Purchased line items
CREATE TABLE order_items (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    order_id INTEGER NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
    store_sku TEXT NOT NULL,
    quantity INTEGER NOT NULL DEFAULT 1,
    unit_price DECIMAL(10, 2) NOT NULL
);

-- SKU translation bridge between Storefront and CJ Dropshipping
CREATE TABLE sku_mappings (
    store_sku TEXT PRIMARY KEY,
    cj_pid TEXT NOT NULL,
    cj_vid TEXT NOT NULL,
    cj_variant_name TEXT,
    wholesale_cost DECIMAL(10, 2) NOT NULL,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

When curating catalog products, we populate sku_mappings with CJ's official vid and wholesale price. When the fulfillment daemon queries unfulfilled orders, it performs an INNER JOIN against this table. If an order contains even one line item missing an active mapping, the script halts fulfillment for that order, transitions its status to NEEDS_ATTENTION, and alerts our team rather than dispatching an incomplete package.

The automated fulfillment pipeline in Python

With authentication and data modeling established, we built fulfill_orders.py. The script is structured as an idempotent state machine:

1. Query unfulfilled orders from SQLite where status = 'PAID' and cj_order_id IS NULL.
2. Fetch associated line items, resolving each item's supplier vid and unit cost from sku_mappings.
3. Compile the validated shipping address and product array into CJ's Open API 2.0 schema.
4. Execute an HTTP POST to /shopping/order/createOrder.
5. If successful, save CJ's generated order ID, update status to PROCESSING, and commit the transaction.

Here is the production implementation:

import sqlite3
import logging
from typing import List, Dict, Any
import httpx
from cj_auth import CJAuthManager

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("fulfillment.worker")

DB_PATH = "data/nyxeris.db"
API_BASE = "https://developers.cjdropshipping.com/api2.0/v1"

def fetch_pending_orders(conn: sqlite3.Connection) -> List[Dict[str, Any]]:
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("""
        SELECT id, order_number, customer_name, email, phone,
               address_line1, address_line2, city, state, postal_code, country_code
        FROM orders
        WHERE status = 'PAID' AND cj_order_id IS NULL
        ORDER BY id ASC
        LIMIT 20;
    """)
    return [dict(r) for r in cur.fetchall()]

def fetch_order_items(conn: sqlite3.Connection, order_id: int) -> List[Dict[str, Any]]:
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("""
        SELECT i.store_sku, i.quantity, i.unit_price, m.cj_vid, m.wholesale_cost
        FROM order_items i
        LEFT JOIN sku_mappings m ON i.store_sku = m.store_sku
        WHERE i.order_id = ?;
    """, (order_id,))
    return [dict(r) for r in cur.fetchall()]

def process_fulfillment_queue():
    auth = CJAuthManager(api_key="CJ5792999@api@YOUR_CJ_API_KEY")
    token = auth.get_valid_token()

    with sqlite3.connect(DB_PATH) as conn:
        orders = fetch_pending_orders(conn)
        if not orders:
            logger.info("No unfulfilled orders in queue.")
            return

        logger.info(f"Discovered {len(orders)} paid orders waiting for fulfillment.")

        for ord_info in orders:
            order_id = ord_info["id"]
            order_no = ord_info["order_number"]
            line_items = fetch_order_items(conn, order_id)

            unmapped = [item["store_sku"] for item in line_items if not item.get("cj_vid")]
            if unmapped:
                logger.error(f"Order {order_no} contains unmapped SKUs: {unmapped}. Skipping.")
                conn.execute(
                    "UPDATE orders SET status = 'NEEDS_ATTENTION' WHERE id = ?",
                    (order_id,)
                )
                conn.commit()
                continue

            cj_products = [
                {
                    "vid": item["cj_vid"],
                    "quantity": int(item["quantity"]),
                    "unitPrice": float(item["wholesale_cost"])
                }
                for item in line_items
            ]

            payload = {
                "orderNumber": order_no,
                "shippingCustomerName": ord_info["customer_name"],
                "shippingPhone": ord_info["phone"] or "10000000000",
                "shippingAddress": ord_info["address_line1"],
                "shippingAddress2": ord_info["address_line2"] or "",
                "shippingCity": ord_info["city"],
                "shippingProvince": ord_info["state"],
                "shippingCountryCode": ord_info["country_code"].upper(),
                "shippingZip": ord_info["postal_code"],
                "remark": "Nyxeris Automated Fulfillment Bridge",
                "products": cj_products
            }

            headers = {
                "CJ-Access-Token": token,
                "Content-Type": "application/json"
            }

            try:
                with httpx.Client(timeout=25.0) as client:
                    response = client.post(
                        f"{API_BASE}/shopping/order/createOrder",
                        headers=headers,
                        json=payload
                    )
                
                result_data = response.json() if response.status_code == 200 else {}
                
                if response.status_code == 200 and result_data.get("result"):
                    cj_order_id = result_data.get("data")
                    logger.info(f"Successfully created CJ fulfillment ticket {cj_order_id} for order {order_no}")
                    
                    conn.execute("""
                        UPDATE orders
                        SET status = 'PROCESSING', cj_order_id = ?, fulfilled_at = CURRENT_TIMESTAMP
                        WHERE id = ?;
                    """, (cj_order_id, order_id))
                    conn.commit()
                else:
                    error_msg = result_data.get("message") or f"HTTP {response.status_code}: {response.text}"
                    logger.error(f"CJ API submission rejected for order {order_no}: {error_msg}")
                    conn.execute(
                        "UPDATE orders SET status = 'FAILED' WHERE id = ?",
                        (order_id,)
                    )
                    conn.commit()

            except Exception as ex:
                logger.error(f"Network error submitting order {order_no}: {ex}")

if __name__ == "__main__":
    process_fulfillment_queue()

Polling tracking numbers and closing the customer loop

Submitting an order to CJ creates the warehouse fulfillment record, but customer expectations aren't satisfied until they receive a valid shipping notification with carrier tracking.

Depending on whether an order ships from a domestic US hub or an international facility in Shenzhen, CJ's warehouse takes between 12 and 36 hours to pack the hardware, affix shipping labels, and assign a tracking number. To close this loop autonomously, we built a companion script: sync_tracking.py.

The tracking daemon queries all local orders marked PROCESSING where a cj_order_id exists but tracking_number is still null. It queries CJ's order inspection endpoint (/shopping/order/getOrderDetail), extracts the carrier tracking code, and triggers our customer dispatch notification:

import sqlite3
import logging
import httpx
from cj_auth import CJAuthManager

logger = logging.getLogger("tracking.sync")
DB_PATH = "data/nyxeris.db"
API_BASE = "https://developers.cjdropshipping.com/api2.0/v1"

def sync_tracking_updates():
    auth = CJAuthManager(api_key="CJ5792999@api@YOUR_CJ_API_KEY")
    token = auth.get_valid_token()

    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row
        cur = conn.cursor()
        cur.execute("""
            SELECT id, order_number, cj_order_id, email, customer_name
            FROM orders
            WHERE status = 'PROCESSING' AND cj_order_id IS NOT NULL AND tracking_number IS NULL;
        """)
        pending_orders = cur.fetchall()

        if not pending_orders:
            logger.info("No pending orders waiting on tracking numbers.")
            return

        logger.info(f"Checking tracking status for {len(pending_orders)} processing orders...")

        for ord_row in pending_orders:
            cj_id = ord_row["cj_order_id"]
            headers = {"CJ-Access-Token": token}
            params = {"orderId": cj_id}

            try:
                resp = httpx.get(
                    f"{API_BASE}/shopping/order/getOrderDetail",
                    headers=headers,
                    params=params,
                    timeout=15.0
                )
                res_json = resp.json()

                if res_json.get("result") and res_json.get("data"):
                    order_data = res_json["data"]
                    track_num = order_data.get("trackingNumber")
                    carrier = order_data.get("logisticName", "Standard Tracked Delivery")
                    cj_status = order_data.get("orderStatus")

                    if track_num:
                        logger.info(f"Tracking assigned for {ord_row['order_number']}: {track_num} ({carrier})")
                        conn.execute("""
                            UPDATE orders
                            SET status = 'SHIPPED', tracking_number = ?, carrier = ?
                            WHERE id = ?;
                        """, (track_num, carrier, ord_row["id"]))
                        conn.commit()

                        send_dispatch_email(
                            to_email=ord_row["email"],
                            customer_name=ord_row["customer_name"],
                            order_number=ord_row["order_number"],
                            tracking_number=track_num,
                            carrier=carrier,
                            tracking_url=f"https://t.17track.net/en#nums={track_num}"
                        )
            except Exception as e:
                logger.error(f"Error querying tracking for CJ order {cj_id}: {e}")

def send_dispatch_email(to_email, customer_name, order_number, tracking_number, carrier, tracking_url):
    logger.info(f"Sent shipping email to {to_email} for order {order_number} ({tracking_number})")

if __name__ == "__main__":
    sync_tracking_updates()

Both scripts run as background cron jobs on our production server:

# Run fulfillment queue every 30 minutes
*/30 * * * * cd /opt/nyxeris && /opt/nyxeris/venv/bin/python fulfill_orders.py >> /var/log/fulfillment.log 2>&1

# Poll for logistics tracking updates twice daily at 04:00 and 16:00 UTC
0 4,16 * * * cd /opt/nyxeris && /opt/nyxeris/venv/bin/python sync_tracking.py >> /var/log/tracking.log 2>&1

Production edge cases: balance, timeouts, and state mismatches

Deploying API automation against physical supply chains quickly reveals quirks that API documentation rarely prepares you for. During our first fortnight in production, we encountered three specific edge cases that required architectural hardening:

1. The prepaid wallet balance exhaustion trap

Unlike standard consumer checkouts, CJ Dropshipping's API does not charge a linked credit card per order. All automated orders are debited against a prepaid cash balance in your CJ Dropshipping Wallet.

If an order costs $28.50 in product cost and air freight, but your CJ wallet only contains $22.00, the API call does not fail with an HTTP 402 or an error status. Instead, CJ returns result: true, assigns an order ID, and silently places the order into an internal state called PENDING_PAYMENT.

The warehouse will not touch the order until you manually deposit funds. To prevent orders from getting stuck in limbo, we added an automated balance inspection pre-check. If our CJ balance drops below $150, the script dispatches a high-priority Telegram alert directly to my phone before executing the batch.

2. State and province name sanitization

CJ's automated carrier routing engine is exceptionally strict regarding regional subdivisions. For orders bound for the United States or Canada, CJ strictly requires two-letter postal abbreviations (e.g., CA, TX, ON). If a customer types "California" or "Ontario" into your checkout, CJ throws a fatal address validation error.

Conversely, for destinations in Great Britain or Australia, CJ frequently rejects county codes and demands full administrative region names (e.g., "Greater London" rather than "LDN"). We resolved this by placing an address normalization filter directly in the Python pipeline prior to payload serialization:

US_STATE_CODES = {
    "ALABAMA": "AL", "ALASKA": "AK", "ARIZONA": "AZ", "ARKANSAS": "AR",
    "CALIFORNIA": "CA", "COLORADO": "CO", "CONNECTICUT": "CT", "FLORIDA": "FL",
    "GEORGIA": "GA", "HAWAII": "HI", "ILLINOIS": "IL", "NEW YORK": "NY",
    "TEXAS": "TX", "WASHINGTON": "WA"
}

def clean_region_code(state: str, country_code: str) -> str:
    cleaned = state.strip().upper()
    if country_code == "US":
        return US_STATE_CODES.get(cleaned, cleaned)
    return cleaned

3. Network timeouts and duplicate fulfillment protection

If an HTTP socket timeout occurs while CJ's servers are committing a fulfillment order, your client raises an exception. If your script naively retries the batch, you risk creating two duplicate warehouse orders and getting billed twice for the same customer.

Fortunately, CJ's createOrder endpoint enforces unique constraints on the orderNumber parameter. If you re-submit an existing order number, CJ returns an error message containing "Order number already exists". We explicitly catch this response string, query CJ's /shopping/order/list endpoint using our store's order number, recover the existing CJ order ID, update SQLite, and proceed smoothly without human intervention.

Operational lessons from shipping automated hardware

Replacing manual fulfillment with a lightweight Python service completely changed the operational profile of our e-commerce project:

• Zero address typos: Address data flows immutably from Stripe checkout directly into CJ's shipping label generator without human transcription errors.
• Faster fulfillment speed: Orders placed while we sleep are queued and submitted to CJ within thirty minutes, entering morning warehouse picking shifts without delay.
• Zero ongoing SaaS tax: Rather than paying $30 to $80 per month for third-party Shopify apps with rigid limitations, our fulfillment pipeline runs on the same $6/month VPS that hosts our database, costing virtually nothing to operate.

If you run a custom web application or self-hosted storefront, don't let supplier logistics intimidate you into adopting closed SaaS ecosystems. With SQLite, a clean SKU translation table, and 200 lines of Python, you can build an automated, dependable order fulfillment pipeline that scales effortlessly.