diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..0e5369a --- /dev/null +++ b/.env.example @@ -0,0 +1,35 @@ +# Copy to .env and fill in. Never commit real secrets. + +# --- Salesforce auth (choose ONE style) --- +# Style A: username/password + security token +SFDC_USERNAME= +SFDC_PASSWORD= +SFDC_SECURITY_TOKEN= +# Optional connected-app credentials +SFDC_CONSUMER_KEY= +SFDC_CONSUMER_SECRET= +# "login" for production, "test" for a sandbox +SFDC_DOMAIN=login + +# Style B: pre-obtained OAuth access token + instance URL +# SFDC_ACCESS_TOKEN= +# SFDC_INSTANCE_URL=https://your-instance.my.salesforce.com + +# --- What to enrich --- +SFDC_SOBJECT=Account +SFDC_LAST_ENRICHED_FIELD=Last_Enriched__c +SFDC_DOMAIN_FIELD=Website +ENRICH_INTERVAL_DAYS=90 +SFDC_BATCH_SIZE=200 +MAX_RECORDS=0 + +# --- Enrichment provider: "mock" (offline, deterministic) or "http" --- +ENRICHMENT_PROVIDER=mock +ENRICHMENT_API_URL= +ENRICHMENT_API_KEY= +ENRICHMENT_TIMEOUT_SECONDS=30 +# Optional override, e.g. "industry=Industry,employee_count=NumberOfEmployees" +ENRICHMENT_FIELD_MAPPING= + +# --- Run controls --- +DRY_RUN=false diff --git a/.github/workflows/enrich-sfdc.yml b/.github/workflows/enrich-sfdc.yml new file mode 100644 index 0000000..7952bf8 --- /dev/null +++ b/.github/workflows/enrich-sfdc.yml @@ -0,0 +1,67 @@ +name: Enrich Salesforce (every 90 days) + +on: + schedule: + # Runs at 06:00 UTC on the 1st of Jan, Apr, Jul, Oct — i.e. roughly every 90 + # days / quarterly. GitHub cron cannot express a true "every 90 days" period, + # so the job also enforces a per-record 90-day freshness window in SOQL, + # meaning no record is ever re-enriched more often than every 90 days even if + # the job runs more frequently. + - cron: "0 6 1 */3 *" + workflow_dispatch: + inputs: + dry_run: + description: "Run without writing changes back to Salesforce" + type: boolean + default: false + max_records: + description: "Cap number of records processed (0 = no limit)" + type: string + default: "0" + +concurrency: + group: enrich-sfdc + cancel-in-progress: false + +jobs: + enrich: + runs-on: ubuntu-latest + timeout-minutes: 60 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + run: pip install -r requirements.txt + + - name: Run enrichment + env: + PYTHONPATH: src + # --- Salesforce auth (set these as repository/environment secrets) --- + SFDC_USERNAME: ${{ secrets.SFDC_USERNAME }} + SFDC_PASSWORD: ${{ secrets.SFDC_PASSWORD }} + SFDC_SECURITY_TOKEN: ${{ secrets.SFDC_SECURITY_TOKEN }} + SFDC_CONSUMER_KEY: ${{ secrets.SFDC_CONSUMER_KEY }} + SFDC_CONSUMER_SECRET: ${{ secrets.SFDC_CONSUMER_SECRET }} + SFDC_ACCESS_TOKEN: ${{ secrets.SFDC_ACCESS_TOKEN }} + SFDC_INSTANCE_URL: ${{ secrets.SFDC_INSTANCE_URL }} + SFDC_DOMAIN: ${{ vars.SFDC_DOMAIN || 'login' }} + # --- What to enrich --- + SFDC_SOBJECT: ${{ vars.SFDC_SOBJECT || 'Account' }} + SFDC_LAST_ENRICHED_FIELD: ${{ vars.SFDC_LAST_ENRICHED_FIELD || 'Last_Enriched__c' }} + SFDC_DOMAIN_FIELD: ${{ vars.SFDC_DOMAIN_FIELD || 'Website' }} + ENRICH_INTERVAL_DAYS: ${{ vars.ENRICH_INTERVAL_DAYS || '90' }} + # --- Enrichment provider --- + ENRICHMENT_PROVIDER: ${{ vars.ENRICHMENT_PROVIDER || 'mock' }} + ENRICHMENT_API_URL: ${{ vars.ENRICHMENT_API_URL }} + ENRICHMENT_API_KEY: ${{ secrets.ENRICHMENT_API_KEY }} + # --- Run controls --- + DRY_RUN: ${{ inputs.dry_run }} + MAX_RECORDS: ${{ inputs.max_records || '0' }} + run: python -m sfdc_enrichment -v diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..17439aa --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.py[cod] +.env +.venv/ +venv/ +.pytest_cache/ +*.egg-info/ +.coverage +htmlcov/ diff --git a/README.md b/README.md index 2125657..740cb44 100644 --- a/README.md +++ b/README.md @@ -1 +1,127 @@ -# misc \ No newline at end of file +# SFDC Enrichment Workflow + +A small, self-contained workflow that keeps Salesforce (SFDC) records enriched +with firmographic data on a rolling **90-day cadence**. + +Every record is (re-)enriched at most once per 90 days: the job finds records +that were never enriched or whose `Last_Enriched__c` timestamp is older than 90 +days, looks them up via an enrichment provider (by company domain), writes the +attributes back, and stamps `Last_Enriched__c` so the record is left alone until +the next window. + +## How the 90-day cadence works + +GitHub Actions `cron` cannot express a true "every 90 days" period, so the +cadence is enforced in **two complementary layers**: + +1. **Schedule** — the workflow runs quarterly (`0 6 1 */3 *`, i.e. 06:00 UTC on + the 1st of Jan/Apr/Jul/Oct), roughly every 90 days. +2. **Per-record freshness window** — every run only selects records whose + `Last_Enriched__c` is null or older than `ENRICH_INTERVAL_DAYS` (default 90). + This means even if you run the job daily/weekly, no record is re-enriched + more often than every 90 days. This is the more robust interpretation of + "enrich every 90 days" and is safe to run on any schedule. + +## Layout + +``` +.github/workflows/enrich-sfdc.yml # scheduled + manual GitHub Actions workflow +src/sfdc_enrichment/ + config.py # env-driven configuration + salesforce_client.py # SOQL query + bulk update wrapper (simple-salesforce) + enrichment.py # provider interface + mock/http implementations + workflow.py # orchestration + 90-day freshness logic + cli.py # command-line entrypoint +tests/ # pytest suite (no live org required) +``` + +## Prerequisites (Salesforce) + +Add a custom datetime field on the target object (default `Account`) named +`Last_Enriched__c` (API name). The workflow reads and writes this field to track +the 90-day window. You can point at a different object/field via env vars. + +## Configuration + +All configuration comes from environment variables (see `.env.example`). + +| Variable | Default | Description | +| --- | --- | --- | +| `SFDC_USERNAME` / `SFDC_PASSWORD` / `SFDC_SECURITY_TOKEN` | – | Username/password auth | +| `SFDC_ACCESS_TOKEN` / `SFDC_INSTANCE_URL` | – | Token auth (alternative) | +| `SFDC_DOMAIN` | `login` | `login` (prod) or `test` (sandbox) | +| `SFDC_SOBJECT` | `Account` | Object to enrich | +| `SFDC_LAST_ENRICHED_FIELD` | `Last_Enriched__c` | Datetime field tracking enrichment | +| `SFDC_DOMAIN_FIELD` | `Website` | Field used as the enrichment lookup key | +| `ENRICH_INTERVAL_DAYS` | `90` | Re-enrichment window (days) | +| `SFDC_BATCH_SIZE` | `200` | Bulk update batch size | +| `MAX_RECORDS` | `0` | Cap records per run (0 = no limit) | +| `ENRICHMENT_PROVIDER` | `mock` | `mock` (offline) or `http` | +| `ENRICHMENT_API_URL` / `ENRICHMENT_API_KEY` | – | HTTP provider settings | +| `ENRICHMENT_FIELD_MAPPING` | – | e.g. `industry=Industry,employee_count=NumberOfEmployees` | +| `DRY_RUN` | `false` | Enrich but do not write back | + +## Usage + +Install dependencies and run locally: + +```bash +pip install -r requirements.txt +cp .env.example .env # fill in credentials + +# Preview the SOQL that will be used (no connection needed) +PYTHONPATH=src python -m sfdc_enrichment --show-query + +# Dry run: query + enrich but do not write back +PYTHONPATH=src python -m sfdc_enrichment --dry-run -v + +# Real run +PYTHONPATH=src python -m sfdc_enrichment -v +``` + +The command prints a JSON summary, e.g.: + +```json +{ + "candidates": 120, + "enriched": 118, + "skipped_no_data": 2, + "updated": 118, + "failed": 0, + "dry_run": false, + "errors": [] +} +``` + +## GitHub Actions + +The workflow in `.github/workflows/enrich-sfdc.yml` runs on the quarterly +schedule and can be triggered manually (`workflow_dispatch`) with `dry_run` and +`max_records` inputs. + +Set these as repository **secrets** (sensitive) and **variables** (non-secret): + +- Secrets: `SFDC_USERNAME`, `SFDC_PASSWORD`, `SFDC_SECURITY_TOKEN`, + `SFDC_CONSUMER_KEY`, `SFDC_CONSUMER_SECRET`, `SFDC_ACCESS_TOKEN`, + `SFDC_INSTANCE_URL`, `ENRICHMENT_API_KEY` (set only those you use). +- Variables: `SFDC_DOMAIN`, `SFDC_SOBJECT`, `SFDC_LAST_ENRICHED_FIELD`, + `SFDC_DOMAIN_FIELD`, `ENRICH_INTERVAL_DAYS`, `ENRICHMENT_PROVIDER`, + `ENRICHMENT_API_URL`. + +## Enrichment providers + +- **`mock`** — deterministic offline data derived from the domain. Great for + dry runs, local dev, and CI (no API key, no network). +- **`http`** — calls `GET {ENRICHMENT_API_URL}?domain=` with a bearer + token and normalizes common vendor response shapes. To integrate a specific + vendor (Clearbit, ZoomInfo, Apollo, ...) implement `EnrichmentProvider` in + `enrichment.py` and register it in `build_provider`. + +## Development + +```bash +pip install -r requirements-dev.txt +PYTHONPATH=src python -m pytest +``` + +The test suite runs fully offline (no Salesforce org or API key required). diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..eeb9d41 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +addopts = -q diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..13f6026 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest>=8.0.0 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..95618b0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +simple-salesforce>=1.12.6 +requests>=2.31.0 +python-dotenv>=1.0.1 diff --git a/src/sfdc_enrichment/__init__.py b/src/sfdc_enrichment/__init__.py new file mode 100644 index 0000000..19918ca --- /dev/null +++ b/src/sfdc_enrichment/__init__.py @@ -0,0 +1,9 @@ +"""Salesforce (SFDC) enrichment workflow. + +A small, self-contained package that keeps Salesforce records enriched with +firmographic data on a rolling 90-day cadence. +""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/src/sfdc_enrichment/__main__.py b/src/sfdc_enrichment/__main__.py new file mode 100644 index 0000000..bfdcd0c --- /dev/null +++ b/src/sfdc_enrichment/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sfdc_enrichment/cli.py b/src/sfdc_enrichment/cli.py new file mode 100644 index 0000000..716ccf5 --- /dev/null +++ b/src/sfdc_enrichment/cli.py @@ -0,0 +1,88 @@ +"""Command-line entrypoint for the SFDC enrichment workflow.""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys + +from .config import ConfigError, load_config +from .enrichment import build_provider +from .salesforce_client import SalesforceClient +from .workflow import build_candidate_query, run_enrichment + + +def _configure_logging(verbose: bool) -> None: + logging.basicConfig( + level=logging.DEBUG if verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + +def build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="sfdc-enrich", + description="Enrich Salesforce records on a rolling 90-day cadence.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Query and enrich but do not write anything back to Salesforce.", + ) + parser.add_argument( + "--max-records", + type=int, + default=None, + help="Cap the number of records processed in this run (0 = no limit).", + ) + parser.add_argument( + "--show-query", + action="store_true", + help="Print the SOQL that would be used and exit.", + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging.") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_arg_parser().parse_args(argv) + _configure_logging(args.verbose) + + # dotenv is optional; load a local .env if present for convenience. + try: + from dotenv import load_dotenv + + load_dotenv() + except Exception: # pragma: no cover - dotenv is optional + pass + + try: + config = load_config() + if args.dry_run: + config.dry_run = True + if args.max_records is not None: + config.max_records = args.max_records + + if args.show_query: + print(build_candidate_query(config)) + return 0 + + config.salesforce.validate() + sf_client = SalesforceClient(config.salesforce) + provider = build_provider(config.enrichment) + summary = run_enrichment(config, sf_client, provider) + except ConfigError as exc: + print(f"Configuration error: {exc}", file=sys.stderr) + return 2 + except Exception as exc: # noqa: BLE001 + logging.getLogger(__name__).exception("Enrichment run failed") + print(f"Run failed: {exc}", file=sys.stderr) + return 1 + + print(json.dumps(summary.as_dict(), indent=2)) + return 0 if summary.failed == 0 else 1 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/sfdc_enrichment/config.py b/src/sfdc_enrichment/config.py new file mode 100644 index 0000000..ab2f858 --- /dev/null +++ b/src/sfdc_enrichment/config.py @@ -0,0 +1,162 @@ +"""Configuration loading for the SFDC enrichment workflow. + +All configuration is read from environment variables so the workflow can run +identically on a laptop, a cron box, or in GitHub Actions (via secrets). +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + +# How often a single record should be re-enriched. The "every 90 days" cadence +# is enforced per-record via this window rather than relying purely on when the +# job happens to run, so a record is never re-enriched more often than this. +DEFAULT_ENRICH_INTERVAL_DAYS = 90 + + +def _get_bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "y", "on"} + + +def _get_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None or raw.strip() == "": + return default + try: + return int(raw) + except ValueError as exc: # pragma: no cover - defensive + raise ConfigError(f"Environment variable {name} must be an integer, got {raw!r}") from exc + + +class ConfigError(RuntimeError): + """Raised when required configuration is missing or invalid.""" + + +@dataclass +class SalesforceConfig: + """Credentials + connection settings for Salesforce. + + Supports the two most common auth styles: + * Username/password + security token (``password`` grant). + * A pre-obtained ``access_token`` + ``instance_url`` (e.g. from an + external OAuth flow / connected app). + """ + + username: str | None = None + password: str | None = None + security_token: str | None = None + consumer_key: str | None = None + consumer_secret: str | None = None + access_token: str | None = None + instance_url: str | None = None + domain: str = "login" # "login" for prod, "test" for sandbox + + def validate(self) -> None: + has_token_auth = bool(self.access_token and self.instance_url) + has_password_auth = bool(self.username and self.password and self.security_token) + if not (has_token_auth or has_password_auth): + raise ConfigError( + "Salesforce auth is not configured. Provide either " + "SFDC_ACCESS_TOKEN + SFDC_INSTANCE_URL, or " + "SFDC_USERNAME + SFDC_PASSWORD + SFDC_SECURITY_TOKEN." + ) + + +@dataclass +class EnrichmentConfig: + """Settings for the enrichment provider that supplies firmographic data.""" + + provider: str = "mock" # "mock" or "http" + api_url: str | None = None + api_key: str | None = None + timeout_seconds: int = 30 + + +@dataclass +class WorkflowConfig: + """Top-level configuration for a single enrichment run.""" + + salesforce: SalesforceConfig + enrichment: EnrichmentConfig + sobject: str = "Account" + # Field on the sobject that stores when the record was last enriched. + last_enriched_field: str = "Last_Enriched__c" + # The domain/website field used as the enrichment lookup key. + domain_field: str = "Website" + enrich_interval_days: int = DEFAULT_ENRICH_INTERVAL_DAYS + batch_size: int = 200 + max_records: int = 0 # 0 == no limit + dry_run: bool = False + # Mapping of provider-returned keys -> Salesforce field API names. + field_mapping: dict[str, str] = field(default_factory=dict) + + +DEFAULT_FIELD_MAPPING = { + "industry": "Industry", + "employee_count": "NumberOfEmployees", + "annual_revenue": "AnnualRevenue", + "phone": "Phone", +} + + +def load_config() -> WorkflowConfig: + """Build a :class:`WorkflowConfig` from environment variables.""" + + salesforce = SalesforceConfig( + username=os.getenv("SFDC_USERNAME"), + password=os.getenv("SFDC_PASSWORD"), + security_token=os.getenv("SFDC_SECURITY_TOKEN"), + consumer_key=os.getenv("SFDC_CONSUMER_KEY"), + consumer_secret=os.getenv("SFDC_CONSUMER_SECRET"), + access_token=os.getenv("SFDC_ACCESS_TOKEN"), + instance_url=os.getenv("SFDC_INSTANCE_URL"), + domain=os.getenv("SFDC_DOMAIN", "login"), + ) + + enrichment = EnrichmentConfig( + provider=os.getenv("ENRICHMENT_PROVIDER", "mock").lower(), + api_url=os.getenv("ENRICHMENT_API_URL"), + api_key=os.getenv("ENRICHMENT_API_KEY"), + timeout_seconds=_get_int("ENRICHMENT_TIMEOUT_SECONDS", 30), + ) + + field_mapping = dict(DEFAULT_FIELD_MAPPING) + custom_mapping = os.getenv("ENRICHMENT_FIELD_MAPPING") + if custom_mapping: + field_mapping = _parse_field_mapping(custom_mapping) + + return WorkflowConfig( + salesforce=salesforce, + enrichment=enrichment, + sobject=os.getenv("SFDC_SOBJECT", "Account"), + last_enriched_field=os.getenv("SFDC_LAST_ENRICHED_FIELD", "Last_Enriched__c"), + domain_field=os.getenv("SFDC_DOMAIN_FIELD", "Website"), + enrich_interval_days=_get_int("ENRICH_INTERVAL_DAYS", DEFAULT_ENRICH_INTERVAL_DAYS), + batch_size=_get_int("SFDC_BATCH_SIZE", 200), + max_records=_get_int("MAX_RECORDS", 0), + dry_run=_get_bool("DRY_RUN", False), + field_mapping=field_mapping, + ) + + +def _parse_field_mapping(raw: str) -> dict[str, str]: + """Parse a mapping string of the form ``key=Field,key2=Field2``.""" + + mapping: dict[str, str] = {} + for pair in raw.split(","): + pair = pair.strip() + if not pair: + continue + if "=" not in pair: + raise ConfigError( + f"Invalid ENRICHMENT_FIELD_MAPPING entry {pair!r}; expected 'key=SalesforceField'." + ) + key, sf_field = pair.split("=", 1) + mapping[key.strip()] = sf_field.strip() + if not mapping: + raise ConfigError("ENRICHMENT_FIELD_MAPPING was provided but no valid entries were parsed.") + return mapping diff --git a/src/sfdc_enrichment/enrichment.py b/src/sfdc_enrichment/enrichment.py new file mode 100644 index 0000000..57914e1 --- /dev/null +++ b/src/sfdc_enrichment/enrichment.py @@ -0,0 +1,133 @@ +"""Enrichment providers. + +A provider takes a lookup key (typically a company domain) and returns a flat +dict of firmographic attributes. Implementations are intentionally small and +pluggable so a real vendor (Clearbit, ZoomInfo, Apollo, ...) can be dropped in +by implementing :class:`EnrichmentProvider`. +""" + +from __future__ import annotations + +import hashlib +import logging +from abc import ABC, abstractmethod +from typing import Any + +from .config import EnrichmentConfig, ConfigError + +logger = logging.getLogger(__name__) + + +class EnrichmentProvider(ABC): + """Interface for anything that can enrich a record by domain.""" + + @abstractmethod + def enrich(self, domain: str) -> dict[str, Any]: + """Return firmographic attributes for ``domain``. + + Returns an empty dict when no data is available. + """ + + +class MockEnrichmentProvider(EnrichmentProvider): + """Deterministic offline provider. + + Produces stable, fake-but-plausible data derived from the domain. Useful for + dry runs, local development, and tests without hitting a paid API. + """ + + _INDUSTRIES = [ + "Technology", + "Financial Services", + "Healthcare", + "Manufacturing", + "Retail", + ] + + def enrich(self, domain: str) -> dict[str, Any]: + if not domain: + return {} + digest = hashlib.sha256(domain.encode("utf-8")).hexdigest() + seed = int(digest[:8], 16) + return { + "industry": self._INDUSTRIES[seed % len(self._INDUSTRIES)], + "employee_count": (seed % 5000) + 1, + "annual_revenue": ((seed % 500) + 1) * 100_000, + "phone": f"+1-555-{seed % 1000:03d}-{(seed >> 8) % 10000:04d}", + } + + +class HttpEnrichmentProvider(EnrichmentProvider): + """Generic HTTP enrichment provider. + + Calls ``GET {api_url}?domain={domain}`` with a bearer token and normalizes a + handful of common response shapes into the flat schema the workflow expects. + """ + + def __init__(self, config: EnrichmentConfig, session: Any | None = None) -> None: + if not config.api_url: + raise ConfigError("ENRICHMENT_API_URL is required for the 'http' provider.") + self._config = config + if session is None: + import requests # imported lazily so tests/mock runs need no network deps + + session = requests.Session() + self._session = session + + def enrich(self, domain: str) -> dict[str, Any]: + if not domain: + return {} + headers = {"Accept": "application/json"} + if self._config.api_key: + headers["Authorization"] = f"Bearer {self._config.api_key}" + try: + response = self._session.get( + self._config.api_url, + params={"domain": domain}, + headers=headers, + timeout=self._config.timeout_seconds, + ) + response.raise_for_status() + payload = response.json() + except Exception as exc: # noqa: BLE001 - surface as a soft failure + logger.warning("Enrichment lookup failed for %s: %s", domain, exc) + return {} + return self._normalize(payload) + + @staticmethod + def _normalize(payload: dict[str, Any]) -> dict[str, Any]: + if not isinstance(payload, dict): + return {} + # Support a couple of common vendor field names. + def pick(*keys: str) -> Any: + for key in keys: + if key in payload and payload[key] not in (None, ""): + return payload[key] + return None + + result: dict[str, Any] = {} + industry = pick("industry", "category", "sector") + employees = pick("employee_count", "employees", "metrics_employees") + revenue = pick("annual_revenue", "revenue", "estimated_annual_revenue") + phone = pick("phone", "phone_number") + + if industry is not None: + result["industry"] = industry + if employees is not None: + result["employee_count"] = employees + if revenue is not None: + result["annual_revenue"] = revenue + if phone is not None: + result["phone"] = phone + return result + + +def build_provider(config: EnrichmentConfig) -> EnrichmentProvider: + """Factory that returns the configured provider.""" + + provider = config.provider.lower() + if provider == "mock": + return MockEnrichmentProvider() + if provider == "http": + return HttpEnrichmentProvider(config) + raise ConfigError(f"Unknown enrichment provider {config.provider!r}. Use 'mock' or 'http'.") diff --git a/src/sfdc_enrichment/salesforce_client.py b/src/sfdc_enrichment/salesforce_client.py new file mode 100644 index 0000000..d4cd1f7 --- /dev/null +++ b/src/sfdc_enrichment/salesforce_client.py @@ -0,0 +1,64 @@ +"""Thin wrapper around the Salesforce REST API. + +Isolates all Salesforce-specific behavior (auth, SOQL, bulk-ish updates) behind +a small interface so the workflow logic stays testable without a live org. +""" + +from __future__ import annotations + +import logging +from typing import Any, Iterable + +from .config import SalesforceConfig + +logger = logging.getLogger(__name__) + + +class SalesforceClient: + """Wraps :mod:`simple_salesforce` with the operations the workflow needs.""" + + def __init__(self, config: SalesforceConfig, connection: Any | None = None) -> None: + config.validate() + self._config = config + self._sf = connection if connection is not None else self._connect(config) + + @staticmethod + def _connect(config: SalesforceConfig) -> Any: + from simple_salesforce import Salesforce # lazy import + + if config.access_token and config.instance_url: + logger.info("Connecting to Salesforce with a pre-obtained access token.") + return Salesforce( + instance_url=config.instance_url, + session_id=config.access_token, + ) + logger.info("Connecting to Salesforce with username/password auth.") + return Salesforce( + username=config.username, + password=config.password, + security_token=config.security_token, + consumer_key=config.consumer_key, + consumer_secret=config.consumer_secret, + domain=config.domain, + ) + + def query_all(self, soql: str) -> list[dict[str, Any]]: + """Run a SOQL query and return all records (following pagination).""" + + logger.debug("SOQL: %s", soql) + result = self._sf.query_all(soql) + return list(result.get("records", [])) + + def update_records(self, sobject: str, records: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + """Update records in bulk. + + Each record dict must include an ``Id`` key plus the fields to update. + Returns the per-record results from the Bulk API. + """ + + records = list(records) + if not records: + return [] + handler = getattr(self._sf.bulk, sobject) + logger.info("Updating %d %s record(s) via the Bulk API.", len(records), sobject) + return handler.update(records) diff --git a/src/sfdc_enrichment/workflow.py b/src/sfdc_enrichment/workflow.py new file mode 100644 index 0000000..65dc560 --- /dev/null +++ b/src/sfdc_enrichment/workflow.py @@ -0,0 +1,184 @@ +"""Orchestration for the SFDC enrichment workflow. + +High level flow for a single run: + +1. Find records that are stale (never enriched, or last enriched > N days ago) + and that have a usable lookup key (a website/domain). +2. Enrich each record via the configured provider. +3. Write the enriched attributes back, stamping ``Last_Enriched__c`` so the same + record is not touched again until the next 90-day window. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any +from urllib.parse import urlparse + +from .config import WorkflowConfig +from .enrichment import EnrichmentProvider +from .salesforce_client import SalesforceClient + +logger = logging.getLogger(__name__) + + +@dataclass +class RunSummary: + """Result of a single enrichment run.""" + + candidates: int = 0 + enriched: int = 0 + skipped_no_data: int = 0 + updated: int = 0 + failed: int = 0 + dry_run: bool = False + errors: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "candidates": self.candidates, + "enriched": self.enriched, + "skipped_no_data": self.skipped_no_data, + "updated": self.updated, + "failed": self.failed, + "dry_run": self.dry_run, + "errors": self.errors, + } + + +def normalize_domain(value: str | None) -> str: + """Turn a website/URL field into a bare domain suitable for lookups.""" + + if not value: + return "" + value = value.strip() + if not value: + return "" + if "://" not in value: + value = "//" + value + netloc = urlparse(value).netloc or "" + netloc = netloc.split("@")[-1] # drop any credentials + netloc = netloc.split(":")[0] # drop any port + if netloc.startswith("www."): + netloc = netloc[4:] + return netloc.lower() + + +def build_candidate_query(config: WorkflowConfig, now: datetime | None = None) -> str: + """Build the SOQL that selects records due for (re)enrichment.""" + + now = now or datetime.now(timezone.utc) + cutoff = now - timedelta(days=config.enrich_interval_days) + cutoff_literal = cutoff.strftime("%Y-%m-%dT%H:%M:%SZ") + + domain_field = config.domain_field + last_field = config.last_enriched_field + + select_fields = ["Id", domain_field, last_field] + select_clause = ", ".join(dict.fromkeys(select_fields)) # de-dupe, keep order + + where = ( + f"{domain_field} != null " + f"AND ({last_field} = null OR {last_field} < {cutoff_literal})" + ) + soql = f"SELECT {select_clause} FROM {config.sobject} WHERE {where}" + if config.max_records and config.max_records > 0: + soql += f" LIMIT {config.max_records}" + return soql + + +def _build_update( + record: dict[str, Any], + enriched: dict[str, Any], + config: WorkflowConfig, + now: datetime, +) -> dict[str, Any]: + """Map provider output onto Salesforce field API names for one record.""" + + update: dict[str, Any] = {"Id": record["Id"]} + for provider_key, sf_field in config.field_mapping.items(): + if provider_key in enriched and enriched[provider_key] not in (None, ""): + update[sf_field] = enriched[provider_key] + update[config.last_enriched_field] = now.strftime("%Y-%m-%dT%H:%M:%SZ") + return update + + +def run_enrichment( + config: WorkflowConfig, + sf_client: SalesforceClient, + provider: EnrichmentProvider, + now: datetime | None = None, +) -> RunSummary: + """Execute one enrichment pass and return a :class:`RunSummary`.""" + + now = now or datetime.now(timezone.utc) + summary = RunSummary(dry_run=config.dry_run) + + soql = build_candidate_query(config, now=now) + records = sf_client.query_all(soql) + summary.candidates = len(records) + logger.info("Found %d candidate %s record(s) due for enrichment.", summary.candidates, config.sobject) + + updates: list[dict[str, Any]] = [] + for record in records: + domain = normalize_domain(record.get(config.domain_field)) + if not domain: + summary.skipped_no_data += 1 + continue + try: + enriched = provider.enrich(domain) + except Exception as exc: # noqa: BLE001 - one bad record shouldn't kill the run + summary.failed += 1 + summary.errors.append(f"{record.get('Id')}: enrichment error: {exc}") + logger.warning("Enrichment failed for %s (%s): %s", record.get("Id"), domain, exc) + continue + + if not enriched: + summary.skipped_no_data += 1 + continue + + summary.enriched += 1 + updates.append(_build_update(record, enriched, config, now)) + + if not updates: + logger.info("No records to update.") + return summary + + if config.dry_run: + logger.info("[dry-run] Would update %d record(s); skipping write.", len(updates)) + for update in updates: + logger.debug("[dry-run] %s", update) + return summary + + _apply_updates(config, sf_client, updates, summary) + return summary + + +def _apply_updates( + config: WorkflowConfig, + sf_client: SalesforceClient, + updates: list[dict[str, Any]], + summary: RunSummary, +) -> None: + """Write updates back to Salesforce in batches, tracking results.""" + + batch_size = max(1, config.batch_size) + for start in range(0, len(updates), batch_size): + batch = updates[start : start + batch_size] + try: + results = sf_client.update_records(config.sobject, batch) + except Exception as exc: # noqa: BLE001 + summary.failed += len(batch) + summary.errors.append(f"batch update error: {exc}") + logger.error("Batch update failed: %s", exc) + continue + + for result in results or []: + if result.get("success"): + summary.updated += 1 + else: + summary.failed += 1 + errs = result.get("errors") or [] + summary.errors.append(f"{result.get('id')}: {errs}") diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d956842 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,4 @@ +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..6d7562c --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,57 @@ +import pytest + +from sfdc_enrichment.config import ( + ConfigError, + _parse_field_mapping, + load_config, +) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + for key in list(__import__("os").environ): + if key.startswith(("SFDC_", "ENRICH", "DRY_RUN", "MAX_RECORDS")): + monkeypatch.delenv(key, raising=False) + + +def test_load_config_defaults(): + config = load_config() + assert config.sobject == "Account" + assert config.enrich_interval_days == 90 + assert config.enrichment.provider == "mock" + assert config.field_mapping["industry"] == "Industry" + + +def test_load_config_reads_env(monkeypatch): + monkeypatch.setenv("SFDC_SOBJECT", "Lead") + monkeypatch.setenv("ENRICH_INTERVAL_DAYS", "30") + monkeypatch.setenv("DRY_RUN", "true") + monkeypatch.setenv("MAX_RECORDS", "10") + config = load_config() + assert config.sobject == "Lead" + assert config.enrich_interval_days == 30 + assert config.dry_run is True + assert config.max_records == 10 + + +def test_salesforce_validate_requires_auth(): + config = load_config() + with pytest.raises(ConfigError): + config.salesforce.validate() + + +def test_salesforce_validate_token_auth(monkeypatch): + monkeypatch.setenv("SFDC_ACCESS_TOKEN", "tok") + monkeypatch.setenv("SFDC_INSTANCE_URL", "https://na1.salesforce.com") + config = load_config() + config.salesforce.validate() # should not raise + + +def test_parse_field_mapping(): + mapping = _parse_field_mapping("industry=Industry, employees=NumberOfEmployees") + assert mapping == {"industry": "Industry", "employees": "NumberOfEmployees"} + + +def test_parse_field_mapping_invalid(): + with pytest.raises(ConfigError): + _parse_field_mapping("bogus") diff --git a/tests/test_enrichment.py b/tests/test_enrichment.py new file mode 100644 index 0000000..e7b063b --- /dev/null +++ b/tests/test_enrichment.py @@ -0,0 +1,87 @@ +import pytest + +from sfdc_enrichment.config import EnrichmentConfig, ConfigError +from sfdc_enrichment.enrichment import ( + HttpEnrichmentProvider, + MockEnrichmentProvider, + build_provider, +) + + +def test_mock_provider_is_deterministic(): + provider = MockEnrichmentProvider() + a = provider.enrich("example.com") + b = provider.enrich("example.com") + assert a == b + assert set(a) == {"industry", "employee_count", "annual_revenue", "phone"} + + +def test_mock_provider_empty_domain(): + assert MockEnrichmentProvider().enrich("") == {} + + +def test_build_provider_mock(): + provider = build_provider(EnrichmentConfig(provider="mock")) + assert isinstance(provider, MockEnrichmentProvider) + + +def test_build_provider_unknown(): + with pytest.raises(ConfigError): + build_provider(EnrichmentConfig(provider="nope")) + + +def test_http_provider_requires_url(): + with pytest.raises(ConfigError): + HttpEnrichmentProvider(EnrichmentConfig(provider="http")) + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self._payload + + +class _FakeSession: + def __init__(self, payload): + self._payload = payload + self.calls = [] + + def get(self, url, params=None, headers=None, timeout=None): + self.calls.append({"url": url, "params": params, "headers": headers}) + return _FakeResponse(self._payload) + + +def test_http_provider_normalizes_response(): + session = _FakeSession( + {"category": "Software", "employees": 42, "revenue": 1_000_000, "phone_number": "+1-555"} + ) + provider = HttpEnrichmentProvider( + EnrichmentConfig(provider="http", api_url="https://api.example.com/enrich", api_key="k"), + session=session, + ) + result = provider.enrich("acme.com") + assert result == { + "industry": "Software", + "employee_count": 42, + "annual_revenue": 1_000_000, + "phone": "+1-555", + } + assert session.calls[0]["params"] == {"domain": "acme.com"} + assert session.calls[0]["headers"]["Authorization"] == "Bearer k" + + +def test_http_provider_handles_errors_gracefully(): + class _BoomSession: + def get(self, *a, **k): + raise RuntimeError("network down") + + provider = HttpEnrichmentProvider( + EnrichmentConfig(provider="http", api_url="https://api.example.com/enrich"), + session=_BoomSession(), + ) + assert provider.enrich("acme.com") == {} diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 0000000..db9abca --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,141 @@ +from datetime import datetime, timezone + +import pytest + +from sfdc_enrichment.config import ( + DEFAULT_FIELD_MAPPING, + EnrichmentConfig, + SalesforceConfig, + WorkflowConfig, +) +from sfdc_enrichment.enrichment import MockEnrichmentProvider +from sfdc_enrichment.workflow import ( + build_candidate_query, + normalize_domain, + run_enrichment, +) + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("https://www.Example.com/path?q=1", "example.com"), + ("http://foo.co.uk", "foo.co.uk"), + ("bar.io", "bar.io"), + ("www.baz.com", "baz.com"), + ("user:pass@secure.com:8443", "secure.com"), + ("", ""), + (None, ""), + (" ", ""), + ], +) +def test_normalize_domain(raw, expected): + assert normalize_domain(raw) == expected + + +def _config(**overrides): + base = dict( + salesforce=SalesforceConfig(access_token="tok", instance_url="https://x.my.salesforce.com"), + enrichment=EnrichmentConfig(provider="mock"), + field_mapping=dict(DEFAULT_FIELD_MAPPING), + ) + base.update(overrides) + return WorkflowConfig(**base) + + +def test_build_candidate_query_contains_freshness_window(): + config = _config() + now = datetime(2026, 7, 22, 6, 0, 0, tzinfo=timezone.utc) + soql = build_candidate_query(config, now=now) + assert "FROM Account" in soql + assert "Website != null" in soql + assert "Last_Enriched__c = null" in soql + # 90 days before 2026-07-22 is 2026-04-23. + assert "2026-04-23T06:00:00Z" in soql + + +def test_build_candidate_query_respects_max_records(): + config = _config(max_records=5) + soql = build_candidate_query(config) + assert soql.endswith("LIMIT 5") + + +class FakeSalesforceClient: + def __init__(self, records): + self._records = records + self.updated_batches = [] + + def query_all(self, soql): + return list(self._records) + + def update_records(self, sobject, records): + records = list(records) + self.updated_batches.append((sobject, records)) + return [{"success": True, "id": r["Id"], "errors": []} for r in records] + + +def test_run_enrichment_updates_records(): + records = [ + {"Id": "001A", "Website": "https://acme.com", "Last_Enriched__c": None}, + {"Id": "001B", "Website": "https://globex.com", "Last_Enriched__c": None}, + ] + client = FakeSalesforceClient(records) + config = _config() + summary = run_enrichment(config, client, MockEnrichmentProvider()) + + assert summary.candidates == 2 + assert summary.enriched == 2 + assert summary.updated == 2 + assert summary.failed == 0 + # One batch, both records, each with Id + Last_Enriched__c + mapped fields. + assert len(client.updated_batches) == 1 + sobject, batch = client.updated_batches[0] + assert sobject == "Account" + for update in batch: + assert "Id" in update + assert "Last_Enriched__c" in update + assert "Industry" in update + + +def test_run_enrichment_skips_records_without_domain(): + records = [{"Id": "001C", "Website": "", "Last_Enriched__c": None}] + client = FakeSalesforceClient(records) + summary = run_enrichment(_config(), client, MockEnrichmentProvider()) + assert summary.candidates == 1 + assert summary.skipped_no_data == 1 + assert summary.updated == 0 + assert client.updated_batches == [] + + +def test_run_enrichment_dry_run_does_not_write(): + records = [{"Id": "001D", "Website": "acme.com", "Last_Enriched__c": None}] + client = FakeSalesforceClient(records) + summary = run_enrichment(_config(dry_run=True), client, MockEnrichmentProvider()) + assert summary.enriched == 1 + assert summary.updated == 0 + assert summary.dry_run is True + assert client.updated_batches == [] + + +def test_run_enrichment_batches_updates(): + records = [ + {"Id": f"00{i}", "Website": f"company{i}.com", "Last_Enriched__c": None} + for i in range(5) + ] + client = FakeSalesforceClient(records) + summary = run_enrichment(_config(batch_size=2), client, MockEnrichmentProvider()) + assert summary.updated == 5 + assert len(client.updated_batches) == 3 # 2 + 2 + 1 + + +def test_run_enrichment_handles_provider_errors(): + class BoomProvider: + def enrich(self, domain): + raise RuntimeError("boom") + + records = [{"Id": "001E", "Website": "acme.com", "Last_Enriched__c": None}] + client = FakeSalesforceClient(records) + summary = run_enrichment(_config(), client, BoomProvider()) + assert summary.failed == 1 + assert summary.updated == 0 + assert summary.errors