How to Enrich a Databricks Table With LeadOcean (Scheduled Notebook Job)

export LEADOCEAN_API_KEY="paste-your-key-here"

CREATE TABLE IF NOT EXISTS main.sales.leads (
  id            BIGINT GENERATED ALWAYS AS IDENTITY,
  linkedin_url  STRING,
  email         STRING,
  enriched_at   TIMESTAMP,
  enrich_status STRING,   -- matched | not_found
  work_email    STRING,
  email_status  STRING,
  phone         STRING,
  profile       STRING
);

CREATE TABLE IF NOT EXISTS main.sales.companies (
  id          BIGINT GENERATED ALWAYS AS IDENTITY,
  domain      STRING,
  enriched_at TIMESTAMP,
  company     STRING
);

INSERT INTO main.sales.leads (linkedin_url)
VALUES ('https://www.linkedin.com/in/janedoe-example');
INSERT INTO main.sales.companies (domain) VALUES ('acme.com');

databricks secrets create-scope leadocean
databricks secrets put-secret leadocean api_key --string-value "$LEADOCEAN_API_KEY"

import json, time, requests
from datetime import datetime, timezone

KEY = dbutils.secrets.get("leadocean", "api_key")
API = "https://api.leadocean.io"
H = {"x-api-key": KEY, "Content-Type": "application/json"}

class Stop(Exception):
    pass

def call(path, body):
    time.sleep(0.25)  # stays under 5 requests a second on Free
    r = requests.post(API + path, headers=H, json=body, timeout=30)
    if r.status_code == 404:
        return None
    if r.status_code in (401, 402):
        raise Stop(f"stop:{r.status_code}")
    if r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", "1")))
        return call(path, body)
    r.raise_for_status()
    return r.json().get("data")

now = datetime.now(timezone.utc)
people, cos = [], []

try:
    pending = spark.table("main.sales.leads").where("enriched_at IS NULL").limit(50).collect()
    for row in pending:
        key = {"linkedin_url": row.linkedin_url} if row.linkedin_url else {"email": row.email}
        d = call("/v1/people/enrich", {**key, "reveal_email": True})
        if d is None:
            people.append((row.id, "not_found", None, None, None, None))
            continue
        c = d.get("contact_data") or {}
        people.append((
            row.id, "matched",
            (c.get("contact_emails") or [{}])[0].get("email"),
            c.get("email_status"),
            (c.get("contact_phones") or [{}])[0].get("phone"),
            json.dumps(d.get("profile_data")),
        ))

    for row in spark.table("main.sales.companies").where("enriched_at IS NULL").limit(50).collect():
        d = call("/v1/companies/enrich", {"domain": row.domain})
        cos.append((row.id, json.dumps(d) if d else None))
except Stop as e:
    print(e)  # bad key or records spent: write what we have, then end

cols = "id long, enrich_status string, work_email string, email_status string, phone string, profile string"
if people:
    spark.createDataFrame(people, cols).createOrReplaceTempView("people_updates")
    spark.sql(f"""
      MERGE INTO main.sales.leads t USING people_updates s ON t.id = s.id
      WHEN MATCHED THEN UPDATE SET t.enrich_status = s.enrich_status,
        t.work_email = s.work_email, t.email_status = s.email_status,
        t.phone = s.phone, t.profile = s.profile,
        t.enriched_at = timestamp'{now:%Y-%m-%d %H:%M:%S}'
    """)
if cos:
    spark.createDataFrame(cos, "id long, company string").createOrReplaceTempView("company_updates")
    spark.sql(f"""
      MERGE INTO main.sales.companies t USING company_updates s ON t.id = s.id
      WHEN MATCHED THEN UPDATE SET t.company = s.company,
        t.enriched_at = timestamp'{now:%Y-%m-%d %H:%M:%S}'
    """)
print({"people": len(people), "companies": len(cos)})

Trigger type:  Scheduled
Schedule type: Advanced
Cron:          0 0 * * * ?     (every hour, on the hour)
Time zone:     UTC

SELECT id, work_email, email_status, phone, enrich_status
FROM main.sales.leads
WHERE enrich_status = 'matched'
  AND email_status IN ('verified', 'catch_all_valid', 'catch_all');

{"name":"email_status","total":13,"matched":13,"values":["verified","catch_all_valid","catch_all","risky","unknown","untested","invalid","role","disposable","spam_trap","abuse","derived","none"],"truncated":false}

{
  "success": true,
  "data": {
    "profile_data": { "...": "name, title, company" },
    "contact_data": {
      "has_email": true,
      "email_status": "verified",
      "contact_emails": [{ "email": "jane.doe@acme.com", "type": "work" }],
      "contact_phones": [{ "phone": "+14155550133", "type": "mobile" }]
    }
  }
}