Coveo Indexing Pipeline Architecture: Push API, Stream API, Document Extensions, and Security Identity Providers

1. Ingestion Modes: Push API vs. Stream API vs. Crawlers

Coveo supports three primary ingress mechanisms for enterprise data:

2. The Document Processing Pipeline (DPP) Lifecycle

Once a document payload is received via the API or crawler, it enters the Document Processing Pipeline (DPP):

1. Format Conversion & Text Extraction: The document is parsed (supporting over 40+ formats including HTML, PDF, Word, XML, and JSON) to extract raw text, clean whitespace, and preserve document structure.
2. Python Indexing Extension Execution: Custom Python scripts intercept the document in-flight to enrich attributes, scrape custom DOM tags, calculate business scores, or drop irrelevant entries.
3. Field Schema Mapping: Extracted keys are mapped to Coveo field definitions with strict typing (Integer, Float, Date, String, Vector, and Facet flags).
4. Security Descriptor Binding: The document is bound to an explicit Access Control List (ACL) resolving allowed and denied entities.
5. Index Shard Replication: The document is written to Lucene segments and distributed across multi-region read replicas.

3. Deep-Dive: Python Document Extension Scripts

One of Coveo's most powerful architectural features is Indexing Extensions: sandboxed Python 3 scripts that execute during the DPP lifecycle. Extensions allow you to manipulate metadata before indexing occurs.

Common use cases include:

• Dynamic Margin Scoring: Calculating gross profit margins in real-time to allow margin-weighted search ranking.
• Text Cleansing & PII Redaction: Masking Social Security Numbers, credit card patterns, or personal email addresses before index commit.
• Synthetic Hierarchies: Building multi-level taxonomy strings (e.g. Electronics;Laptops;Gaming) for hierarchical facet widgets.

Production Example: Catalog Enrichment Extension

# Coveo Indexing Extension (Python 3)
# Executed in Document Processing Pipeline post-conversion

import json
import re

def compute_margin(retail_price, cost):
    try:
        r = float(retail_price)
        c = float(cost)
        if r <= 0:
            return 0.0
        return round(((r - c) / r) * 100, 2)
    except (ValueError, TypeError):
        return 0.0

try:
    # 1. Access document metadata dictionary
    meta = document.get_meta_data()

    # 2. Extract pricing fields
    retail_price = meta.get("retail_price", [None])[0]
    cost_price = meta.get("wholesale_cost", [None])[0]

    # 3. Calculate and attach synthetic margin field
    margin = compute_margin(retail_price, cost_price)
    document.add_meta_data({"margin_percent": margin})

    # 4. Enforce PII redaction on product descriptions
    raw_desc = document.get_body_text()
    if raw_desc:
        # Redact potential email addresses or phone patterns
        sanitized = re.sub(r'[\w\.-]+@[\w\.-]+', '[REDACTED]', raw_desc)
        document.set_body_text(sanitized)

    # 5. Conditionally reject discontinued and zero-inventory items
    inventory_count = int(meta.get("inventory_available", [0])[0])
    is_discontinued = meta.get("status", [""])[0].lower() == "discontinued"

    if is_discontinued and inventory_count <= 0:
        document.reject("Product is discontinued and has zero warehouse inventory.")

except Exception as e:
    # Log to Coveo extension console without breaking the pipeline
    log(f"Extension processing error on document {document.uri}: {str(e)}")

4. Enterprise Access Control: Early Binding vs. Late Binding

In enterprise search, returning a search result that the user is not authorized to view is an immediate compliance violation. Coveo implements Early Binding Document-Level Security (DLS):

• Late Binding (Query Time): The search engine retrieves 1,000 candidate documents, then queries Active Directory or Salesforce to filter out unauthorized records. This creates massive latency bottlenecks and distorts facet counts.
• Early Binding (Index Time): During document ingestion, the exact security descriptors (SIDs for Users, Groups, Roles, and Virtual Deny groups) are stamped directly onto the document index core. At query time, the user's authenticated token groups are merged into the query filter via boolean bitset operations in sub-millisecond time.

5. High-Throughput Push API Batch Ingestion Script

Below is a production-grade Python script for bulk pushing catalog batches to Coveo Push API v2:

import requests
import json

COVEO_ORG_ID = "myenterpriseorg"
SOURCE_ID = "source_crm_kb_v2"
API_KEY = "xx_coveo_push_api_key"

ENDPOINT = f"https://api.cloud.coveo.com/push/v1/organizations/{COVEO_ORG_ID}/sources/{SOURCE_ID}/documents/batch"

def push_catalog_batch(documents):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "addOrUpdate": [
            {
                "documentId": doc["uri"],
                "title": doc["title"],
                "body": doc["content"],
                "data": json.dumps(doc),
                "permissions": [
                    {
                        "allowAnonymous": False,
                        "allowedPermissions": [
                            {"additionalInfo": {}, "identity": "AllEmployees", "identityType": "Group"}
                        ],
                        "deniedPermissions": []
                    }
                ],
                "price": doc["price"],
                "sku": doc["sku"],
                "category": doc["category"]
            }
            for doc in documents
        ],
        "delete": []
    }

    resp = requests.put(ENDPOINT, headers=headers, json=payload)
    if resp.status_code == 200:
        print(f"Batch of {len(documents)} pushed successfully.")
    else:
        raise Exception(f"Push failed: {resp.status_code} - {resp.text}")