Build a Company Enrichment Pipeline (Code Examples Included)

Your SaaS collects user emails at signup. That's all you know about them. But from that email, you can automatically determine their company name, industry, size, funding stage, tech stack, and even what growth signals they're showing right now. All it takes is one API call. This company enrichment API tutorial walks you through building a complete enrichment pipeline — from domain extraction to automated workflows — with production-ready code examples in Python and Node.js.

If you've ever wondered how to enrich company data using an API and turn anonymous signups into fully profiled accounts, you're in the right place. We'll cover every step of the process, including the edge cases most tutorials skip.

Why Company Enrichment Matters for SaaS

Most SaaS products treat every new signup the same way: a generic onboarding flow, a drip email sequence, and maybe a product tour. But here's the problem — a solo founder at a seed-stage startup and a VP of Engineering at a 2,000-person enterprise have fundamentally different needs, budgets, and buying processes. A company data enrichment pipeline lets you distinguish between them the moment they sign up.

Here's what enrichment unlocks:

Personalized onboarding: Route enterprise users to a white-glove experience and startups to self-serve. According to HubSpot's research, personalized onboarding increases activation rates by 20–30%.
Smarter lead scoring: Combine firmographic data (company size, funding) with behavioral data (feature usage, page views) for a lead score that actually predicts conversion. Check out our guide on signal-based lead scoring for a deeper dive.
Automated segmentation: Tag users as SMB, mid-market, or enterprise without manual research. Feed segments into your CRM, marketing automation, and product analytics tools.
Better product analytics: Understand which company types get the most value. If 80% of your power users come from Series B fintech companies, that's a data point your product team needs.
Sales-assisted motion: When enrichment reveals a high-value account, automatically notify your sales team. No more leads slipping through the cracks.

The ROI is clear: enrichment turns a single data point (an email address) into a complete company profile that powers every downstream decision. Let's build it.

Architecture Overview

Before writing any code, let's map out the full SaaS enrichment pipeline architecture. The flow is straightforward:

User signs up with email
    → Extract company domain from email
    → Check if domain is a personal email provider (skip if yes)
    → Call enrichment API with domain
    → Parse and map response to your data model
    → Store enriched data in your database
    → Trigger downstream workflows (onboarding, scoring, alerts)
    → Subscribe to webhooks for real-time signal updates

There are two approaches to implementing this:

Synchronous (inline): Enrich during the signup request. The user waits an extra 200–500ms, but enriched data is available immediately for personalized onboarding. Best when enrichment directly affects the first-run experience.
Asynchronous (background job): Queue the enrichment call as a background task. The user gets a fast signup experience, and enriched data populates within seconds. Best for high-traffic applications or when enrichment powers workflows that don't need to be instant. This is what AWS recommends for event-driven architectures.

For most SaaS products, the async approach is the right call. Queue the enrichment, process it in the background, and let webhooks handle ongoing updates. Here's how to build each component.

Step 1 — Extract Company Domain from Email

The first step in any company data enrichment guide is parsing the email address to extract the company domain. This sounds trivial, but the edge cases matter.

# enrichment/domain_extractor.py

# Common free email providers — enrichment won't return useful data for these
FREE_EMAIL_DOMAINS = {
    "gmail.com", "yahoo.com", "hotmail.com", "outlook.com",
    "aol.com", "icloud.com", "mail.com", "protonmail.com",
    "zoho.com", "yandex.com", "gmx.com", "fastmail.com",
    "tutanota.com", "hey.com", "pm.me", "live.com",
    "msn.com", "me.com", "mac.com"
}

def extract_company_domain(email: str) -> str | None:
    """
    Extract company domain from email address.
    Returns None if the email uses a free/personal provider.
    """
    if not email or "@" not in email:
        return None

    domain = email.strip().lower().split("@")[1]

    # Skip free email providers
    if domain in FREE_EMAIL_DOMAINS:
        return None

    # Skip disposable email domains (optional: use a larger list or API)
    if domain.endswith(".temporary.email") or domain.endswith(".guerrillamail.com"):
        return None

    return domain


# Usage
domain = extract_company_domain("[email protected]")
# Returns: "stripe.com"

domain = extract_company_domain("[email protected]")
# Returns: None (free email, skip enrichment)

A few important notes:

Maintain a comprehensive free email domain list. The example above covers the most common ones, but in production you'll want 300+ domains. The dev.to community maintains several open-source lists.
Consider checking for disposable email domains too. Users signing up with throwaway emails rarely convert.
Some companies use Google Workspace or Microsoft 365 with their own domain — those are not free emails and should be enriched normally.

Step 2 — Call the BounceWatch Enrichment API

Now for the core of the pipeline: making the API call to enrich user data with company information. BounceWatch's API returns firmographic data, funding history, team information, and real-time growth signals in a single request.

Python Example

# enrichment/api_client.py
import requests
from typing import Optional

BOUNCEWATCH_API_BASE = "https://api.bouncewatch.com/api/v1"

class BounceWatchClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json"
        })

    def enrich_company(self, domain: str, modules: list = None) -> Optional[dict]:
        """
        Enrich a company by domain.
        Modules: business, funding, team, signals, technology
        """
        if modules is None:
            modules = ["business", "funding", "team", "signals"]

        try:
            response = self.session.get(
                f"{BOUNCEWATCH_API_BASE}/company/{domain}",
                params={"enrich": ",".join(modules)},
                timeout=10
            )

            if response.status_code == 200:
                return response.json()
            elif response.status_code == 404:
                # Company not found in database
                return None
            elif response.status_code == 429:
                # Rate limited — implement backoff
                retry_after = int(response.headers.get("Retry-After", 60))
                raise RateLimitError(f"Rate limited. Retry after {retry_after}s")
            else:
                response.raise_for_status()

        except requests.exceptions.Timeout:
            raise EnrichmentError("API request timed out")
        except requests.exceptions.ConnectionError:
            raise EnrichmentError("Could not connect to BounceWatch API")

class RateLimitError(Exception):
    pass

class EnrichmentError(Exception):
    pass

Node.js Example

// enrichment/apiClient.js
const axios = require('axios');

const BOUNCEWATCH_API_BASE = 'https://api.bouncewatch.com/api/v1';

class BounceWatchClient {
  constructor(apiKey) {
    this.client = axios.create({
      baseURL: BOUNCEWATCH_API_BASE,
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Accept': 'application/json'
      },
      timeout: 10000
    });
  }

  async enrichCompany(domain, modules = ['business', 'funding', 'team', 'signals']) {
    try {
      const response = await this.client.get(`/company/${domain}`, {
        params: { enrich: modules.join(',') }
      });
      return response.data;

    } catch (error) {
      if (error.response) {
        if (error.response.status === 404) return null;
        if (error.response.status === 429) {
          const retryAfter = error.response.headers['retry-after'] || 60;
          throw new Error(`Rate limited. Retry after ${retryAfter}s`);
        }
      }
      throw new Error(`Enrichment failed: ${error.message}`);
    }
  }
}

module.exports = BounceWatchClient;

Sample API Response

Here's what a typical enrichment response looks like. This is the kind of data you'll get back from a single call — firmographics, funding, team details, and active signals:

{
  "domain": "notion.so",
  "company": {
    "name": "Notion",
    "legal_name": "Notion Labs, Inc.",
    "industry": "Software / Productivity",
    "sub_industry": "Collaboration Tools",
    "employee_count": 800,
    "employee_range": "501-1000",
    "founded_year": 2013,
    "country": "US",
    "city": "San Francisco",
    "description": "All-in-one workspace for notes, tasks, wikis, and databases.",
    "website": "https://www.notion.so"
  },
  "funding": {
    "total_funding_usd": 343000000,
    "last_round_type": "Series C",
    "last_round_amount_usd": 275000000,
    "last_round_date": "2024-11-15",
    "funding_stage": "Late Stage",
    "investors": ["Sequoia Capital", "Index Ventures", "Coatue Management"]
  },
  "team": {
    "total_team_size": 812,
    "engineering_percentage": 42,
    "recent_hires_30d": 18,
    "open_positions": 34,
    "key_people": [
      {"name": "Ivan Zhao", "title": "CEO & Co-founder"},
      {"name": "Simon Last", "title": "CTO & Co-founder"}
    ]
  },
  "signals": [
    {"type": "hiring_surge", "score": 78, "detail": "+18 hires in 30 days"},
    {"type": "expanding", "score": 65, "detail": "New office in London"},
    {"type": "product_launch", "score": 82, "detail": "Launched AI features Q4 2024"}
  ]
}

This single response gives you everything you need to segment, score, and route this account — no manual research required. Compare this with legacy approaches in our API comparison guide.

Step 3 — Map API Response to Your Data Model

Raw API responses shouldn't be stored as-is. Map the fields you need to your internal data model. This keeps your schema clean and decouples your app from the API's structure.

# enrichment/mapper.py

def map_enrichment_to_company(api_response: dict) -> dict:
    """
    Map BounceWatch API response to our internal company model.
    Only extract the fields we actually use.
    """
    if not api_response:
        return {}

    company_data = api_response.get("company", {})
    funding_data = api_response.get("funding", {})
    team_data = api_response.get("team", {})
    signals = api_response.get("signals", [])

    return {
        # Core firmographics
        "company_name": company_data.get("name"),
        "domain": api_response.get("domain"),
        "industry": company_data.get("industry"),
        "sub_industry": company_data.get("sub_industry"),
        "employee_count": company_data.get("employee_count"),
        "employee_range": company_data.get("employee_range"),
        "country": company_data.get("country"),
        "city": company_data.get("city"),
        "founded_year": company_data.get("founded_year"),

        # Funding
        "funding_stage": funding_data.get("funding_stage"),
        "total_funding_usd": funding_data.get("total_funding_usd"),
        "last_round_type": funding_data.get("last_round_type"),
        "last_round_date": funding_data.get("last_round_date"),

        # Team dynamics
        "team_size": team_data.get("total_team_size"),
        "open_positions": team_data.get("open_positions"),
        "recent_hires_30d": team_data.get("recent_hires_30d"),

        # Signals (store as JSON array)
        "active_signals": [
            {
                "type": s.get("type"),
                "score": s.get("score"),
                "detail": s.get("detail")
            }
            for s in signals
        ],

        # Computed fields
        "is_enterprise": company_data.get("employee_count", 0) > 200,
        "is_recently_funded": funding_data.get("last_round_date", "") > "2025-01-01",
        "enrichment_source": "bouncewatch",
        "enriched_at": datetime.utcnow().isoformat()
    }


def save_enrichment(user_id: int, enriched_data: dict, db_session):
    """
    Save enriched company data linked to the user.
    Creates or updates the company record.
    """
    from models import Company, User

    domain = enriched_data.get("domain")
    if not domain:
        return

    # Upsert company record
    company = db_session.query(Company).filter_by(domain=domain).first()
    if company:
        for key, value in enriched_data.items():
            if key != "domain" and value is not None:
                setattr(company, key, value)
    else:
        company = Company(**enriched_data)
        db_session.add(company)

    # Link user to company
    user = db_session.query(User).get(user_id)
    if user:
        user.company_id = company.id

    db_session.commit()
    return company

The key design decisions here:

Upsert logic: Multiple users from the same company should share one company record. Match on domain.
Computed fields: Pre-calculate is_enterprise and is_recently_funded at enrichment time. These boolean flags make downstream queries and workflow conditions much simpler.
Signals as JSON: Store active signals as a JSON column. They change frequently, and you don't need to query individual signals relationally.
Timestamp tracking: Always store enriched_at so you know when data was last refreshed.

Step 4 — Implement Webhook for Real-Time Updates

Enrichment at signup gives you a snapshot. But company data changes — they raise funding, start hiring, open new offices, launch products. Webhooks let you receive these funding signals, hiring signals, and expansion signals in real time without polling.

Flask Webhook Endpoint (Python)

# webhooks/signal_receiver.py
from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["BOUNCEWATCH_WEBHOOK_SECRET"]

def verify_webhook_signature(payload: bytes, signature: str) -> bool:
    """Verify the webhook came from BounceWatch."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.route("/webhooks/bouncewatch/signals", methods=["POST"])
def handle_signal_webhook():
    # Verify signature
    signature = request.headers.get("X-BounceWatch-Signature", "")
    if not verify_webhook_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401

    payload = request.json
    event_type = payload.get("event")  # e.g., "signal.detected", "signal.updated"
    domain = payload.get("domain")
    signal = payload.get("signal")

    if event_type == "signal.detected":
        # New signal detected for a company we're tracking
        handle_new_signal(domain, signal)
    elif event_type == "signal.updated":
        # Existing signal score changed
        handle_signal_update(domain, signal)

    return jsonify({"received": True}), 200

def handle_new_signal(domain: str, signal: dict):
    """Process a new signal — update DB and trigger workflows."""
    company = Company.query.filter_by(domain=domain).first()
    if not company:
        return

    # Append new signal to active_signals
    signals = company.active_signals or []
    signals.append({
        "type": signal["type"],
        "score": signal["score"],
        "detail": signal["detail"],
        "detected_at": signal["detected_at"]
    })
    company.active_signals = signals
    company.enriched_at = datetime.utcnow()
    db.session.commit()

    # Trigger workflows based on signal type
    if signal["type"] == "hiring_surge" and signal["score"] > 70:
        notify_sales_team(company, signal)
    if signal["type"] == "recently_funded":
        recalculate_lead_score(company)

Express Webhook Endpoint (Node.js)

// webhooks/signalReceiver.js
const express = require('express');
const crypto = require('crypto');
const router = express.Router();

const WEBHOOK_SECRET = process.env.BOUNCEWATCH_WEBHOOK_SECRET;

function verifySignature(payload, signature) {
  const expected = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(payload)
    .digest('hex');
  return crypto.timingSafeEqual(
    Buffer.from(`sha256=${expected}`),
    Buffer.from(signature)
  );
}

router.post('/webhooks/bouncewatch/signals', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-bouncewatch-signature'] || '';

  if (!verifySignature(req.body, signature)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const payload = JSON.parse(req.body);
  const { event, domain, signal } = payload;

  if (event === 'signal.detected') {
    handleNewSignal(domain, signal);
  }

  res.json({ received: true });
});

module.exports = router;

Security note: Always verify webhook signatures. Without verification, anyone could POST fake signal data to your endpoint. The HMAC-SHA256 approach shown above is industry standard — you'll find similar patterns in Stripe, GitHub, and Slack webhook implementations.

Step 5 — Build Automated Workflows on Top of Enrichment

Enrichment data sitting in a database is only useful if it drives decisions. Here are three production workflows that automate company data handling and deliver immediate ROI.

Workflow 1 — Personalized Onboarding

Route new signups to different onboarding paths based on their company profile. This is the highest-impact use case for enrichment — it affects the user's very first experience.

# workflows/onboarding.py

def determine_onboarding_flow(enriched_data: dict) -> str:
    """
    Route users to the appropriate onboarding flow based on company enrichment.
    Returns: 'enterprise', 'growth', 'startup', or 'default'
    """
    employee_count = enriched_data.get("employee_count", 0)
    funding_stage = enriched_data.get("funding_stage", "")
    total_funding = enriched_data.get("total_funding_usd", 0)

    # Enterprise: 200+ employees or $50M+ funding
    if employee_count > 200 or total_funding > 50_000_000:
        return "enterprise"

    # Growth-stage: Series A/B, 50-200 employees
    if funding_stage in ("Series A", "Series B") and employee_count > 50:
        return "growth"

    # Early-stage: Seed/Pre-seed or small teams
    if funding_stage in ("Seed", "Pre-Seed", "Angel") or employee_count < 50:
        return "startup"

    return "default"


def apply_onboarding(user_id: int, flow: str):
    """Apply the onboarding flow to the user."""
    flows = {
        "enterprise": {
            "welcome_template": "enterprise_welcome",
            "assign_csm": True,
            "show_demo_cta": True,
            "trial_length_days": 30,
            "features_unlocked": ["sso", "api_access", "custom_reports"]
        },
        "growth": {
            "welcome_template": "growth_welcome",
            "assign_csm": False,
            "show_demo_cta": True,
            "trial_length_days": 14,
            "features_unlocked": ["api_access", "team_seats"]
        },
        "startup": {
            "welcome_template": "startup_welcome",
            "assign_csm": False,
            "show_demo_cta": False,
            "trial_length_days": 14,
            "features_unlocked": ["basic"]
        }
    }

    config = flows.get(flow, flows["startup"])
    # Apply config to user's onboarding state
    update_user_onboarding(user_id, config)

Workflow 2 — Lead Scoring

Combine enrichment data with behavioral signals for a lead score that actually predicts conversion. This is where signal-based lead scoring really shines — you're not guessing, you're scoring based on real company dynamics.

# workflows/lead_scoring.py

def calculate_lead_score(enriched_data: dict, behavioral_data: dict) -> int:
    """
    Calculate a composite lead score (0-100) from enrichment + behavioral data.
    """
    score = 0

    # === Firmographic scoring (max 40 points) ===
    employee_count = enriched_data.get("employee_count", 0)
    if employee_count > 500:
        score += 15
    elif employee_count > 100:
        score += 12
    elif employee_count > 20:
        score += 8

    funding_stage = enriched_data.get("funding_stage", "")
    funding_scores = {
        "Series C": 15, "Series B": 13, "Series A": 10,
        "Late Stage": 15, "Seed": 5, "Pre-Seed": 2
    }
    score += funding_scores.get(funding_stage, 0)

    # Recently funded = high intent to spend
    if enriched_data.get("is_recently_funded"):
        score += 10

    # === Signal scoring (max 30 points) ===
    signals = enriched_data.get("active_signals", [])
    for signal in signals:
        signal_type = signal.get("type")
        signal_score = signal.get("score", 0)

        if signal_type == "hiring_surge" and signal_score > 60:
            score += 10  # Growing team = growing budget
        if signal_type == "recently_funded":
            score += 10  # Fresh capital = buying mode
        if signal_type == "expanding":
            score += 5   # New markets = new tools needed
        if signal_type == "product_launch":
            score += 5   # Active development = tech spend

    # === Behavioral scoring (max 30 points) ===
    if behavioral_data.get("visited_pricing_page"):
        score += 10
    if behavioral_data.get("api_calls_count", 0) > 50:
        score += 10
    if behavioral_data.get("invited_team_members"):
        score += 10

    return min(score, 100)

A lead with a score above 70 — say a recently funded Series B company with a hiring surge whose user just visited your pricing page — is a hot prospect that your sales team should contact immediately.

Workflow 3 — Sales Alerts

When enrichment or signal webhooks reveal a high-value account, don't let it sit in a database. Push a notification to your sales team in real time.

# workflows/sales_alerts.py
import requests

SLACK_WEBHOOK_URL = os.environ["SLACK_SALES_WEBHOOK_URL"]

def notify_sales_team(company: dict, trigger: str, lead_score: int):
    """
    Send a Slack notification when a high-value account is detected.
    """
    signals_text = ", ".join(
        f"{s['type']} ({s['score']})" for s in company.get("active_signals", [])
    )

    message = {
        "blocks": [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": f"🔥 Hot Lead Detected: {company['company_name']}"}
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Domain:* {company['domain']}"},
                    {"type": "mrkdwn", "text": f"*Industry:* {company['industry']}"},
                    {"type": "mrkdwn", "text": f"*Size:* {company['employee_count']} employees"},
                    {"type": "mrkdwn", "text": f"*Funding:* {company['funding_stage']}"},
                    {"type": "mrkdwn", "text": f"*Lead Score:* {lead_score}/100"},
                    {"type": "mrkdwn", "text": f"*Active Signals:* {signals_text}"},
                    {"type": "mrkdwn", "text": f"*Trigger:* {trigger}"}
                ]
            }
        ]
    }

    requests.post(SLACK_WEBHOOK_URL, json=message)


# Example: Trigger after enrichment
if lead_score > 70 and enriched_data.get("is_enterprise"):
    notify_sales_team(enriched_data, "Enterprise signup + high score", lead_score)

This is the kind of integration that turns your Signal Tracker data into revenue. No more manually reviewing new signups — the system tells your sales team exactly who to call and why.

Step 6 — Handle Edge Cases and Optimize

A production enrichment pipeline needs more than happy-path code. Here are the optimizations that separate a tutorial project from a real system.

Caching with Redis

Don't call the enrichment API every time a user from the same company signs up. Cache results by domain with a 24-hour TTL. Redis is the standard choice here.

# enrichment/cache.py
import redis
import json

redis_client = redis.Redis(host="localhost", port=6379, db=0)
CACHE_TTL = 86400  # 24 hours

def get_cached_enrichment(domain: str) -> dict | None:
    """Check Redis cache before hitting the API."""
    cached = redis_client.get(f"enrichment:{domain}")
    if cached:
        return json.loads(cached)
    return None

def cache_enrichment(domain: str, data: dict):
    """Store enrichment result in Redis with TTL."""
    redis_client.setex(
        f"enrichment:{domain}",
        CACHE_TTL,
        json.dumps(data)
    )

def enrich_with_cache(client: BounceWatchClient, domain: str) -> dict | None:
    """Enrichment with cache-first strategy."""
    # Check cache first
    cached = get_cached_enrichment(domain)
    if cached:
        return cached

    # Cache miss — call API
    result = client.enrich_company(domain)
    if result:
        cache_enrichment(domain, result)

    return result

Rate Limiting with Exponential Backoff

# enrichment/resilience.py
import time
import random

def enrich_with_backoff(client: BounceWatchClient, domain: str, max_retries: int = 3) -> dict | None:
    """Call enrichment API with exponential backoff on rate limits."""
    for attempt in range(max_retries):
        try:
            return client.enrich_company(domain)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait_time)
    return None

Bulk Enrichment for Existing Users

If you're adding enrichment to an existing product, you'll need to backfill data for your current user base. Process in batches to stay within rate limits.

# enrichment/bulk.py
import time

def bulk_enrich_existing_users(db_session, client: BounceWatchClient, batch_size: int = 50):
    """
    Enrich all existing users who haven't been enriched yet.
    Processes in batches with rate limit awareness.
    """
    unenriched_users = (
        db_session.query(User)
        .filter(User.company_id.is_(None))
        .filter(User.email.isnot(None))
        .all()
    )

    total = len(unenriched_users)
    enriched_count = 0
    skipped_count = 0

    for i in range(0, total, batch_size):
        batch = unenriched_users[i:i + batch_size]

        for user in batch:
            domain = extract_company_domain(user.email)
            if not domain:
                skipped_count += 1
                continue

            try:
                result = enrich_with_cache(client, domain)
                if result:
                    mapped = map_enrichment_to_company(result)
                    save_enrichment(user.id, mapped, db_session)
                    enriched_count += 1
            except RateLimitError:
                time.sleep(60)  # Wait and retry
            except Exception as e:
                print(f"Failed to enrich {domain}: {e}")

        # Respect rate limits between batches
        time.sleep(2)

    print(f"Bulk enrichment complete: {enriched_count} enriched, {skipped_count} skipped, {total} total")

Data Freshness Strategy

Webhook-driven updates: Real-time signal changes push to your endpoint automatically.
Monthly re-enrichment: Run a batch job to refresh all company data monthly. Employee counts change, new funding rounds happen, leadership changes occur.
On-demand re-enrichment: Let users trigger a refresh from your UI when they need the latest data.

Complete Pipeline Code

Here's the full pipeline brought together — from email to enriched profile — in a single, production-ready function. As Real Python emphasizes, clean and readable code is maintainable code.

# enrichment/pipeline.py
"""
Complete company enrichment pipeline.
Enrich a user's company data from their email address.
"""
import os
from datetime import datetime

# Initialize client
client = BounceWatchClient(api_key=os.environ["BOUNCEWATCH_API_KEY"])

def run_enrichment_pipeline(user_id: int, email: str, db_session) -> dict:
    """
    Full enrichment pipeline: email → domain → API → map → store → workflows.
    Returns enrichment result or empty dict if not applicable.
    """
    # Step 1: Extract domain
    domain = extract_company_domain(email)
    if not domain:
        return {"status": "skipped", "reason": "personal_email"}

    # Step 2: Check cache, then call API
    api_response = enrich_with_cache(client, domain)
    if not api_response:
        return {"status": "not_found", "domain": domain}

    # Step 3: Map to internal model
    enriched_data = map_enrichment_to_company(api_response)

    # Step 4: Save to database
    company = save_enrichment(user_id, enriched_data, db_session)

    # Step 5: Trigger workflows
    onboarding_flow = determine_onboarding_flow(enriched_data)
    apply_onboarding(user_id, onboarding_flow)

    lead_score = calculate_lead_score(
        enriched_data,
        behavioral_data=get_user_behavior(user_id)
    )

    # Alert sales for high-value accounts
    if lead_score > 70:
        notify_sales_team(enriched_data, "High-score signup", lead_score)

    return {
        "status": "enriched",
        "company": enriched_data.get("company_name"),
        "domain": domain,
        "onboarding_flow": onboarding_flow,
        "lead_score": lead_score,
        "signals_count": len(enriched_data.get("active_signals", []))
    }


# Usage — call this from your signup handler
result = run_enrichment_pipeline(
    user_id=new_user.id,
    email=new_user.email,
    db_session=db.session
)
print(f"Enrichment result: {result}")
# Output: {'status': 'enriched', 'company': 'Notion', 'domain': 'notion.so',
#          'onboarding_flow': 'enterprise', 'lead_score': 82, 'signals_count': 3}

That's the complete pipeline. From a single email address, you now have a fully enriched company profile powering personalized onboarding, lead scoring, and sales alerts — all automated.

Cost Optimization Tips

Enrichment APIs charge per call, so being smart about usage directly impacts your unit economics. Here's how to keep costs low while maintaining data quality:

Use modular enrichment: BounceWatch lets you request specific modules (business, funding, signals, etc.). If you only need firmographics for segmentation, don't request the full signals package. Request only what each workflow needs — this is a key difference highlighted in our Clearbit alternatives comparison.
Cache aggressively: A 24-hour TTL on Redis means multiple signups from the same company only cost you one API call. For large companies, this saves dozens of calls per day.
Batch existing users: Use the bulk enrichment endpoint instead of individual calls when backfilling. Bulk pricing is typically 40–60% cheaper than individual lookups.
Use webhooks instead of polling: Don't re-enrich daily to check for signal changes. Set up webhooks and let BounceWatch push updates to you. This alone can reduce API calls by 90%+.
Skip personal emails early: The free email domain check in Step 1 costs you zero API calls. For a B2B SaaS with 30% consumer signups, that's a 30% cost reduction right there.
Tiered enrichment: For free-tier users, enrich only firmographics. For paying customers or high-engagement users, pull the full dataset including signals. Match API spend to user value.

Explore the full range of available data points in our company database and compare pricing across providers in our BounceWatch vs Apollo comparison.

Start Building Your Enrichment Pipeline Today

You now have everything you need to build a production-grade company enrichment pipeline — from domain extraction to automated workflows. The code examples above are production-ready: add your API key, configure your database models, and you're live.

Here's what to do next:

Get your BounceWatch API key — sign up at bouncewatch.com/api and start with 100 free enrichment calls.
Start with Step 1 and Step 2 — domain extraction and a single API call. You'll see enriched data in minutes.
Add workflows incrementally — start with onboarding routing, then add lead scoring and sales alerts as you see results.
Set up webhooks — once you're enriching at signup, add the webhook endpoint to receive real-time signal updates.

The difference between a SaaS that treats every signup the same and one that instantly understands each user's context is a single enrichment API call. Build the pipeline, ship it today, and watch your conversion metrics improve.

Get your free BounceWatch API key →

How to Build a Company Enrichment Pipeline for Your SaaS (with Code Examples)