SOP-OB-003: ERP Integration Build Checklist

Version: 1.1 Owner: Engineering Lead Audience: Engineering (ENG) Used by: SOP-OB-002 Step A2

1. Purpose and Scope

This checklist is the engineering reference for building a new ERP data integration. It covers every layer of the stack: TypeScript registry, Python Dagster pipeline (connector, orchestrator, transformers), Instance Explorer (capability discovery), factory registrations, Temporal write-back workflows, and testing.

This checklist is also the PR acceptance criteria. A PR for a new integration is not mergeable until all applicable items are checked.

Reference Implementations

Study these before starting:

ERP	Status	Best for studying
Prophet21	GA (most complete)	`sources/prophet21/`, `implementations/p21_sync_orchestrator_v2.py`, `transformers/products.py`
NetSuite	Beta	`sources/netsuite/`, `implementations/netsuite_sync_orchestrator.py` — clean OAuth2 example
Acumatica	In-progress	`sources/acumatica/` — use when ERP requires instance capability probing

All paths are relative to apps/dagster/erp_pipeline/ unless otherwise noted.

2. Phase 0: API Discovery and Scoping

Deliverable: docs/erp/<slug>-discovery.md + field mapping worksheet

3. Phase 1: TypeScript Registry Entry

File: packages/integrations/src/registry/registry.ts

Add a new entry to the INTEGRATIONS record:

<slug>: {
  slug: '<slug>',
  name: '<Display Name>',
  category: 'erp',
  authType: '<basic | oauth2 | api_key | token>',
  capabilities: {
    source: true,
    destination: false,
    actions: false,   // set true only if order write-back is in scope
  },
  status: 'beta',    // always start as beta; promote to active after first live UAT
  configSchema: {
    fields: [
      // One object per credential field shown in the connection UI
      // Example: { key: 'apiUrl', label: 'API URL', type: 'url', required: true }
    ],
  },
},

Slug is lowercase with underscores, no hyphens, no spaces
Slug matches the Python erp_type string exactly (grep ERPSourceFactory to verify)
authType matches the actual auth mechanism (determines which credential fields are shown in the UI)
actions: true only if order write-back is planned (even if not built yet)
status: 'beta' — never set to 'active' before the first live customer UAT passes

4. Phase 2: Python Auth Handler

File: sources/<slug>/auth.py (ERP-specific) + sources/erp_auth.py (shared getter)

4.1 Credential Retrieval

Implement credential retrieval from Infisical via CredentialService — follow the get_p21_credentials pattern in erp_auth.py
Add get_<slug>_credentials(connection_id: str) -> dict to sources/erp_auth.py
The function must return a typed dict with all fields the orchestrator and client need
Handle missing credentials explicitly: raise CredentialNotFoundError with the connection ID, do not return None silently

4.2 Token Management (OAuth2 only)

Implement token acquisition using the appropriate OAuth2 flow (client credentials for service-to-service)
Implement token caching — store in memory for the orchestrator’s lifetime; do not re-fetch on every API call
Implement token expiry detection: check expires_in from the token response; refresh proactively 60 seconds before expiry
Implement token refresh: reuse the same client credentials to obtain a new token
Raise AuthenticationError on auth failure — do not silently fall back or retry with wrong credentials

5. Phase 3: Python HTTP Client

File: sources/<slug>/client.py

Use httpx (preferred) — async client if the orchestrator is async, sync otherwise
Set default timeouts: timeout=httpx.Timeout(connect=10.0, read=30.0, write=30.0, pool=10.0)
Implement retry with exponential backoff for 429 (rate limit) and 5xx responses:
- Max 3 retries, base delay 2s, max delay 30s
- On 429: honor Retry-After header if present
Implement pagination as a generator — yield pages one at a time; do not buffer all results in memory
Log at DEBUG level: method, URL, status code, response time (do not log credentials or response bodies)
Return typed data classes or plain dicts from client methods — do not let httpx.Response objects leak out of the client layer
Handle common ERP error patterns: distinguish auth errors (401/403), not-found (404), and server errors (5xx)
If SOAP/XML: implement an XML serializer/deserializer using lxml or xmltodict; parse to dict before returning

6. Phase 4: DLT Source

File: sources/<slug>/dlt_source.py

Implement a @dlt.source decorated function that returns a DltSource
Implement one @dlt.resource function per entity type (products, customers, etc.)
For incremental sync: use dlt.sources.incremental on the cursor field (e.g., lastModifiedDate)
Set write_disposition="merge" with the ERP’s natural primary key as primary_key
Set _sync_source metadata field to "<slug>" on every record
Add resources: list[str] parameter to allow selective sync (caller passes only requested entity types)
Handle empty result sets gracefully: yield nothing, do not raise exceptions
Yield raw ERP records from the DLT source — transformation is done in the transformer layer, not here
Add a full_sync: bool = False parameter: when True, ignore the incremental cursor and pull all records

7. Phase 5: Transformers

Directory: transformers/ Files: <slug>_products.py, <slug>_customers.py, <slug>_shipping_addresses.py, etc.

For each entity type the ERP supports, implement a transformer:

Function signature: transform_<slug>_<entity>(record: dict) -> dict
Map every ERP-specific field to the normalized Iceberg schema field name
- Reference: transformers/products.py and transformers/customers.py for Prophet21 (the canonical mapping)
- Reference: packages/integrations/src/sync/implementations/prophet21-sync-service.ts for the expected downstream field names
Set _sync_source = "<slug>"
Set delete_flag = False for active records; handle ERP soft-delete flags if the ERP supports them
Set last_synced_at = datetime.now(timezone.utc).isoformat()
Handle None and missing optional fields with sensible defaults: empty string "" for text, None for nullable fields, False for boolean flags
Call reorder_dict_to_schema(record, "<entity_type>") as the final step — required for DLT merge operations with Iceberg
Document ERP-specific quirks in inline comments (e.g., “P21 uses empty string for null addresses”)

Transformer unit tests (minimum 5 per entity type)

Normal record with all fields populated
Record with all optional fields missing / null
Record with unexpected extra fields (should be ignored gracefully)
Date format edge case (ERP’s format vs. ISO 8601)
Unicode / special character in a name field

8. Phase 6: Transformer Registry

File: transformers/registry.py

Import each transformer function at the top of the relevant registration block

For each entity type, call:

TransformerRegistry.register_<entity>_transformer("<slug>", transform_<slug>_<entity>)

Add a comment block above the registrations (follow the # === Register Prophet21 Transformers === pattern)
After registrations, verify: assert "<slug>" in TransformerRegistry.list_registered_erp_types() — add this to the integration tests

9. Phase 7: Sync Orchestrator

File: implementations/<slug>_sync_orchestrator.py

10. Phase 8: Instance Explorer

File: sources/<slug>/explorer.py

The Instance Explorer is a first-class deliverable for every new ERP integration. It replaces the manual discovery process (SOP-OB-002 Step A1) for all future customers using this ERP — making onboarding faster and more consistent. The explorer runs once on initial connection and whenever the profile becomes stale (default: 24 hours).

Reference implementation: sources/acumatica/explorer.py — AcumaticaInstanceExplorer with TenantCapabilityProfile Protocol: sources/explorer_protocol.py — BaseExplorer / CapabilityProfile

10.1 Capability Profile Dataclass

Create a <Slug>CapabilityProfile dataclass implementing the CapabilityProfile protocol:

@dataclass
class <Slug>CapabilityProfile:
    erp_type: str = "<slug>"
    profile_version: int = 1
    discovered_at: str = ""
    probe_errors: list[str] = field(default_factory=list)
    # ERP-specific fields:
    api_version: str = ""          # detected version string
    entities: EntityAvailability = field(default_factory=EntityAvailability)
    custom_fields: dict[str, list[str]] = field(default_factory=dict)
    entity_sizes: dict[str, int] = field(default_factory=dict)
    # Add ERP-specific fields as needed (e.g., multi_branch, pricing_engine)

Add to_dict() and from_dict() methods for JSON round-tripping
Add is_stale(max_age_hours: int = 24) -> bool — checks discovered_at age (copy from Acumatica pattern)

10.2 Explorer Class

10.3 Orchestrator Integration

In the sync orchestrator (implementations/<slug>_sync_orchestrator.py), add run_instance_explorer() method:

def run_instance_explorer(self) -> dict:
    explorer = <Slug>InstanceExplorer(self._client)
    profile = explorer.run()
    # Persist to erp_connections.extra_config
    self._save_capability_profile(profile.to_dict())
    return profile.to_dict()

Add _save_capability_profile(profile_dict) — merges profile into erp_connections.extra_config["tenant_capability_profile"] via DB update

10.4 Staleness and Re-probing

The orchestrator’s main sync entrypoint must check is_stale() before each sync run:

profile = TenantCapabilityProfile.from_dict(connection_config.get("tenant_capability_profile", {}))
if profile.is_stale():
    profile = self.run_instance_explorer()

If an endpoint that was previously available starts returning 404 mid-sync, catch it and trigger a re-probe rather than failing the sync

10.5 Explorer Tests

Unit test: probe_entities correctly marks an endpoint as available on HTTP 200 with records, unavailable on 404 or empty list
Unit test: probe_custom_fields correctly identifies custom fields given a sample record dict
Unit test: is_stale() returns True for profiles older than 24h and False for fresh ones
Integration test: full explorer.run() against sandbox ERP returns a profile with probe_errors = []
Integration test: profile entity_sizes matches approximate record counts visible in ERP admin UI

11. Phase 9: Factory Registration

Python Factory

File: sources/erp_source_factory.py

Add "<slug>" to the _SUPPORTED_ERP_TYPES frozenset

Add an elif erp_type == "<slug>": branch in create_orchestrator():

elif erp_type == "<slug>":
    from erp_pipeline.implementations.<slug>_sync_orchestrator import <Slug>SyncOrchestrator
    credentials = get_<slug>_credentials(connection_id)
    connection_config = {
        "api_url": credentials["api_url"],
        # ... all fields the orchestrator needs
    }
    return <Slug>SyncOrchestrator(
        connection_id=connection_id,
        connection_config=connection_config,
    )

Use lazy imports (inside the elif block) to avoid circular dependencies
Update is_implemented() to return True for the new slug only after tests pass

TypeScript Factory

File: packages/integrations/src/sync/erp-sync-factory.ts

If a TS sync service is needed (real-time lookup path): add a case '<slug>': branch returning a new <Slug>SyncService()
If no TS sync service is needed yet (Dagster-only path): add a case '<slug>': branch that throws a descriptive error explaining which layer handles this ERP

12. Phase 10: Default Capabilities

File: apps/dagster/erp_pipeline/capabilities.py

Add "<slug>": ERPCapabilities(...) to DEFAULT_CAPABILITIES
Set each capability based on Phase 0 API discovery findings
Use conservative defaults: disable capabilities that require custom API setup or weren’t confirmed — easier to enable later than to debug a failed sync

Add inline comments explaining why each capability is enabled or disabled:

"<slug>": ERPCapabilities(
    products=True,           # confirmed via /items endpoint
    customers=True,          # confirmed via /accounts endpoint
    cross_references=False,  # endpoint exists but requires add-on license — disabled by default
    shipping_addresses=True, # confirmed via /ship-to endpoint
    ship_to_links=False,     # not available in this ERP's API
),

13. Phase 11: Temporal Write-Back Workflow (Optional — Sandbox Required)

Safety rule (from docs/architecture/integration-architecture-pattern.md): No write operations to any ERP without a confirmed sandbox account. This phase is BLOCKED until sandbox credentials are tested and confirmed.

Confirm sandbox account exists and credentials work (test a write to sandbox, then verify it appeared)
Create workflow file: apps/temporal-worker/workflows/<slug>_submit_order.py (or .ts)
Implement activities following the established safety pattern:
1. fetch_order_from_db(order_id) — read-only, from local DB
2. validate_order_against_erp_<slug>(customer, items) — pure validation, no side effects
3. submit_order_to_<slug>(order_data) — the only activity with ERP write side effects
4. audit_log(workflow_id, operation, status) — write to integration_audit_trail
Implement retry with exponential backoff on activity failures (max 3 retries for submit_order)
Implement compensating transaction / rollback: if submission partially succeeds then fails, cancel or void the partial ERP record before raising the workflow error
All ERP write operations must be logged to integration_audit_trail with: workflow_id, connection_id, operation, request_payload, response_payload, timestamp
Test fully in sandbox before any production enablement
Get explicit product owner written approval before enabling write-back in production for any customer

14. Phase 12: Testing

Unit Tests

Transformer tests: minimum 5 per entity type (see Phase 5 checklist)
Auth handler tests: happy path + token expiry detection + AuthenticationError on bad credentials
HTTP client tests: pagination yields all pages, retry on 429 and 5xx, timeout raises correctly

Integration Tests

Full sync run against sandbox ERP produces records in Iceberg (erp_data_native_v2 dataset)
Record count in Iceberg matches ERP admin UI within 1% for products and customers
Spot-check field mapping: compare 10 sampled records field by field against ERP API response
TransformerRegistry.list_registered_erp_types() includes "<slug>" for all supported entities
ERPSourceFactory.is_implemented("<slug>") returns True
Regression: existing integrations (P21, NetSuite) still pass after changes to shared files

Test Commands

# TypeScript
pnpm --filter @repo/integrations test

# Python
cd apps/dagster && uv run pytest -v

# Specific ERP tests
cd apps/dagster && uv run pytest tests/test_<slug>_integration.py -v

15. Phase 13: Documentation and Handoff

Create docs/erp/<slug>-integration.md with:
- Customer-facing auth setup instructions (what API credentials to obtain and how)
- Required API permissions / roles (specific permission names from vendor docs)
- Known limitations and explicitly unsupported features
- Field mapping table: ERP field → ERP Unlocked normalized field
- Troubleshooting section for the 3–5 most likely setup errors
Update registry status from 'beta' to 'active' only after first live customer UAT passes
Brief CS in writing on limitations before the integration is offered to any new prospect
File PR with label new-erp-integration

16. PR Acceptance Criteria

A PR for a new ERP integration is not mergeable until:

Code completeness

All Phase 0–13 checklist items complete and checked
TypeScript registry entry present with correct auth type and schema
Python connector passes all unit tests (no skipped tests without documented justification)
Full sync against sandbox produces records in Iceberg (completed status in Dagster run log)
TransformerRegistry.list_registered_erp_types() includes new slug for all supported entity types
ERPSourceFactory._SUPPORTED_ERP_TYPES includes new slug
capabilities.py has a DEFAULT_CAPABILITIES entry with inline comments
No hardcoded credentials anywhere in the codebase

Instance Explorer

<Slug>InstanceExplorer exists in sources/<slug>/explorer.py and implements BaseExplorer
<Slug>CapabilityProfile implements CapabilityProfile protocol (has erp_type, discovered_at, profile_version, probe_errors)
run_instance_explorer() exists on the sync orchestrator and persists profile to erp_connections.extra_config
Explorer ran successfully against sandbox: probe_errors is empty or contains only non-critical warnings
is_stale() unit test passing
entity_sizes populated — confirms $count or equivalent works for this ERP

Safety checklist (from integration-architecture-pattern.md)

Sandbox account confirmed and tested (not just claimed)
All write operations use integration_audit_trail
Rollback / compensating transaction documented and tested
Production credentials stored in Infisical, not in env vars or code
Product owner written approval for any write operations before production enablement

Review and handoff

At least one ENG team member (not the author) has reviewed and approved the PR
CS has been briefed on the new ERP, its capabilities, and current limitations
docs/erp/<slug>-integration.md exists and is complete