# bstore_acumatica_to_databricks

This repository contains two deployed components: an incremental Acumatica-to-Databricks ingestion job and a FastAPI Data Validation API. The ingestion job reads ERP records from Acumatica/MYOB Advanced and merges eight streams into Databricks bronze tables. The API provides read-only ERP lookups, warehouse reconciliation, sales and Excel audits, style backfill and deduplication, pipeline health, and operational job controls.

## Business Summary

This service keeps a retailer’s product, supplier, warehouse, purchasing, receiving, and sales data up to date in Databricks. It also gives operations teams tools to check whether ERP data arrived correctly, compare ERP exports with warehouse data, investigate invoices and stock items, repair missing records, and monitor pipeline health.

## Capabilities

| Capability | Category | Status | Access | Exposure |
|---|---|---|---|---|
| ERP-to-Warehouse Data Ingestion | Other | `production` | `execute` | internal, agent_facing |
| Data Reconciliation | Reporting | `production` | `read` | user_facing, agent_facing |
| Data Backfill | Other | `production` | `write` | user_facing, agent_facing |
| Sales Data Auditing | Reporting | `production` | `read` | user_facing, agent_facing |
| Excel-to-Warehouse Data Comparison | Reporting | `production` | `read` | user_facing, agent_facing |
| Invoice Retrieval and Investigation | Orders | `production` | `read` | user_facing, agent_facing |
| Product and Stock Item Lookup | Products | `production` | `read` | user_facing, agent_facing |
| Style Data Validation and Backfill | Products | `production` | `write` | user_facing, agent_facing |
| Style Deduplication | Products | `production` | `destructive` | user_facing, agent_facing |
| Pipeline Operations Monitoring | Reporting | `production` | `read` | user_facing, agent_facing |
| Pipeline Job Orchestration | Other | `production` | `execute` | user_facing, agent_facing |
| Data Validation API Authentication | Authentication | `production` | `read` | user_facing, agent_facing |

**ERP-to-Warehouse Data Ingestion** — Automatically transfers retail purchasing, receiving, sales, product, supplier, and warehouse data from the ERP into the analytics warehouse.

**Data Reconciliation** — Checks whether records in the ERP are present in the Databricks warehouse and identifies missing records.

**Data Backfill** — Restores ERP records that are missing from the warehouse, either for a date range or for specific record keys.

**Sales Data Auditing** — Traces sales orders, invoices, inventory, and totals through the warehouse layers to find missing or inconsistent sales data.

**Excel-to-Warehouse Data Comparison** — Compares Acumatica sales or purchase-receipt spreadsheets with warehouse records to show missing and extra documents.

**Invoice Retrieval and Investigation** — Lets support and operations users look up invoices, their line items, history, and source ERP payloads when investigating sales issues.

**Product and Stock Item Lookup** — Retrieves product details and current stock held at each warehouse from the ERP.

**Style Data Validation and Backfill** — Checks product/style records between Acumatica and Databricks and restores missing product records.

**Style Deduplication** — Finds duplicate product/style records caused by casing or identifier differences and safely rebuilds a cleaned style table.

**Pipeline Operations Monitoring** — Shows whether ingestion streams are fresh and healthy and provides operational run history and lineage information.

**Pipeline Job Orchestration** — Allows authorized operations users or systems to start ingestion, backfill, and stream-specific jobs.

**Data Validation API Authentication** — Restricts validation and operational actions to callers that provide the configured API key.

## Workflows

**Incremental ERP Synchronization** — Moves changed Acumatica records into Databricks bronze tables.

1. Read the current maximum watermark from the target Databricks bronze table, or use the configured initial date.
2. Request changed records from Acumatica in bounded date chunks.
3. Load records into a staging Delta table.
4. MERGE staged records into the target bronze table using configured natural keys.
5. Continue processing other streams when an individual stream fails.

**Reconcile and Backfill Missing ERP Records** — Finds records that exist in Acumatica but not in Databricks and optionally restores them.

1. Select one or more configured streams and a source date range.
2. Read source records from Acumatica and keys from the corresponding warehouse table.
3. Calculate missing keys and return counts and record details.
4. When requested, fetch and merge missing parent and optional child records into bronze.

**Sales or Receipt Export Audit** — Compares a user-provided ERP spreadsheet with warehouse document keys.

1. Upload an Excel sales or purchase-receipt export.
2. Parse the document identifiers and optional date or type filters.
3. Query the corresponding warehouse table.
4. Return missing, extra, matched, and prefix-breakdown results.

**Style Deduplication Apply** — Safely creates a cleaned product/style table after duplicate analysis.

1. Preview duplicate groups and estimated impact.
2. Require an explicit confirmation flag.
3. Optionally validate the generated SQL in dry-run mode.
4. Rebuild the silver style table from bronze with duplicates removed.

**Operational Sync Trigger** — Starts a configured ingestion or backfill job and returns execution tracking details.

1. Validate the requested stream and execution parameters.
2. Trigger the production Azure Container Apps job or configured Databricks job.
3. Return the execution name, job ID, run ID, state, and run page URL.
4. Use operations endpoints to inspect run and stream health.

## Architecture

The repository is a Python monorepo with a standalone command-line ingestion process and a separately deployable FastAPI validation and operations service. The API uses routers, Pydantic request models, service classes, external clients, and parameterized SQL templates. Deployment targets Azure Container Apps for the sync job and validation API, with Databricks Asset Bundles available for scheduled sync execution.

**Components:** acumatica_source.py: Acumatica extraction and source record handling, databricks_dest.py: Databricks SQL staging, table creation, and MERGE loading, sync.py: stream selection and ingestion orchestration, Data_Validation/data_validation/main.py: FastAPI application entry point, Data_Validation/data_validation/api/v1: versioned HTTP routers for audits, reconciliation, operations, ingestion, invoices, stock items, style backfill, and deduplication, Data_Validation/data_validation/services: business logic for audits, reconciliation, backfill, ERP lookups, operations, and deduplication, Data_Validation/data_validation/clients: Acumatica, warehouse, Databricks Jobs, and Azure Container Apps clients, Data_Validation/data_validation/sql: SQL template loading and rendering, Data_Validation/data_validation/models: Pydantic request and response schemas, databricks/resources/jobs.yml and databricks/databricks.yml: Databricks Asset Bundle job deployment, azure-pipelines and bitbucket-pipelines.yml: CI/CD for the sync job and validation API

**Patterns:** Incremental watermark-based extraction, Staging-table then idempotent MERGE loading, Layered API router, service, client, model, and SQL-template architecture, Read-only Acumatica access for lookup and audit operations, Configurable stream definitions loaded from YAML, Explicit confirmation and dry-run protection for destructive table rebuilds, Short-lived shared caching for operations dashboard reads, Separate deployment paths for ingestion and validation components

## Interfaces

| Kind | Identifier | Description |
|---|---|---|
| `http` | `FastAPI /health and /` | Service health and root metadata endpoints. |
| `http` | `FastAPI /api/v1/myob-validation/*` | Reconciliation, missing-record detection, and backfill endpoints for Acumatica-to-Databricks streams. |
| `http` | `FastAPI /api/v1/sales-audit/*` | Sales, inventory, reverse, totals, and Excel-based audit endpoints. |
| `http` | `FastAPI /api/v1/excel-diff/*` | Sales export comparison endpoints. |
| `http` | `FastAPI /api/v1/excel-diff-receipts/*` | Purchase receipt export comparison endpoints. |
| `http` | `FastAPI /api/v1/invoices/*` | Invoice lookup, search, line-item, history, changed, backdated, as-at, and debug operations. |
| `http` | `FastAPI /api/v1/stock-items/*` | Stock item lookup and warehouse stock-on-hand endpoints. |
| `http` | `FastAPI /api/v1/style-backfill/*` | Style validation, reconciliation, and missing-style backfill endpoints. |
| `http` | `FastAPI /api/v1/style-dedup/*` | Style duplicate preview and confirmed apply endpoints. |
| `http` | `FastAPI /api/v1/ops/*` | Stream health, job runs, stream summaries, lineage, job triggering, and backfill orchestration endpoints. |
| `http` | `FastAPI /api/v1/ingestion/*` | Ingestion and direct inspection/debug endpoints. |
| `cli` | `python sync.py [streams...]` | Runs all configured ingestion streams or selected streams such as Vendor, Warehouse, StockItem, SalesInvoice, and PurchaseOrder. |
| `cli` | `Makefile sync-* targets` | Convenience commands for individual stream synchronization, dry runs, deduplication verification, and Databricks bundle deployment. |
| `other` | `Acumatica REST/OData API` | Authenticated source interface used for ERP entities, invoice documents, stock items, and sales-by-brand data. |
| `other` | `Databricks SQL and Jobs APIs` | Warehouse query/write interface and optional job execution interface. |
| `other` | `Azure Container Apps API` | Production orchestration interface used to trigger or inspect the deployed sync job. |

## Data

| Entity | Ownership | Description |
|---|---|---|
| PurchaseOrder | `owns` | Purchase order header records synchronized from Acumatica to the purchase_orders bronze table. |
| PurchaseOrderItems | `owns` | Purchase order line records synchronized to purchase_order_items. |
| PurchaseReceipt | `owns` | Goods-receipt records synchronized to purchase_receipts. |
| SalesInvoice | `owns` | Sales invoice header records synchronized to invoices. |
| SalesInvoiceLineItems | `owns` | Sales invoice line records synchronized to invoices_lines. |
| StockItem | `owns` | Product/style master records synchronized to style and queried for item details and stock. |
| Vendor | `owns` | Supplier records synchronized to supplier. |
| Warehouse | `owns` | Warehouse master records synchronized to warehouse. |
| Invoice | `reads` | AR invoice view queried from Acumatica for lookup and investigation; it is not shown as a separate sync stream. |
| Warehouse Stock Position | `reads` | Per-warehouse QtyOnHand information returned from Acumatica StockItem warehouse details. |
| Sales Audit Layers | `reads` | Bronze, Silver, and Gold invoice, sales-line, style, and final-sales tables used for audit comparisons. |
| Stream Health and Execution Metadata | `reads` | Derived stream freshness, row counts, watermarks, job states, run IDs, and lineage information used by operations views. |
| Excel Audit Upload | `reads` | Transient uploaded sales or purchase-receipt spreadsheet content used for comparison; no durable ownership is established by the evidence. |

## Dependencies

| Name | Kind | Relationship | Criticality |
|---|---|---|---|
| Acumatica / MYOB Advanced ERP | `external_service` | `reads` | `critical` |
| Databricks SQL Warehouse | `database` | `reads` | `critical` |
| Databricks Delta bronze/silver/gold tables | `database` | `writes` | `critical` |
| Databricks Jobs API | `external_service` | `calls` | `required` |
| Azure Container Apps | `external_service` | `calls` | `required` |
| Azure Resource Manager | `external_service` | `authenticates_through` | `required` |
| Azure Key Vault-backed secret scopes | `other` | `uses` | `required` |
| Slack | `external_service` | `publishes` | `optional` |
| FastAPI | `library` | `uses` | `critical` |
| Pydantic | `library` | `uses` | `critical` |
| pandas and Excel parsers | `library` | `uses` | `required` |
| YAML stream configuration | `library` | `uses` | `required` |

## Technology

Python 3.12, FastAPI, Uvicorn, Pydantic v2, Requests, Databricks SQL Connector, Databricks SDK, pandas, python-calamine, openpyxl, PyYAML, SQL templates, Databricks Delta tables, Azure Container Apps Jobs, Databricks Asset Bundles, Docker, Bitbucket Pipelines, Azure Pipelines, pytest, pytest-asyncio, httpx

## Limitations

- The evidence does not establish a user-facing web UI or screens in this repository; the API is described as serving a dashboard maintained elsewhere.
- Acumatica access is implemented as read-only source access. The API's POST endpoints perform validation, backfill, deduplication, or orchestration actions in this service and warehouse, not writes back to Acumatica.
- The exact complete route list and request/response payloads are not available in the supplied source excerpts; route groups are therefore summarized rather than enumerated individually.
- BigQuery appears in repository scanner data, but the supplied authoritative implementation is centered on Databricks and no active BigQuery client is shown. BigQuery should not be treated as a confirmed runtime dependency for this repository.
- The repository contains both the ingestion job and the Data Validation API, which have separate dependencies, containers, deployment triggers, and runtime responsibilities.
- The evidence shows production deployment configuration but does not provide runtime health or proof that every configured stream is currently operational.
- No dedicated persistent audit or execution database is shown; operational freshness is derived from warehouse data and external job APIs.

## Agent Instructions

Use this repository for Acumatica-to-Databricks synchronization, ERP-to-warehouse data quality investigation, product and invoice lookups, reconciliation, backfill, and ingestion operations.

- For normal data movement, use sync.py or the configured operations job rather than directly editing Databricks tables.
- Treat Acumatica as read-only and be cautious with query volume, date ranges, and batch limits.
- For discrepancies, start with reconciliation or Excel comparison, then use invoice, stock-item, or direct invoice-debug endpoints to inspect the ERP source.
- Use style deduplication preview before apply; apply requires explicit confirmation and can rebuild the silver style table.
- Use stream names and target mappings from streams.yaml as the source of truth for reconciliation and backfill requests.
- Do not assume a missing warehouse row means the ERP record is absent; compare against Acumatica and account for stream filters, layer boundaries, and date scope.
- Protected API routes require the configured API key, supplied through the expected X-API-Key mechanism.
- Never expose or request values for credentials, tokens, passwords, API keys, or client secrets; use the existing environment and secret-management configuration.

## Source

| Field | Value |
|---|---|
| Repository | `bstore_acumatica_to_databricks` |
| Branch | `main` |
| Commit | `773cdb33816f` |
| Generated by | `gpt-5.6-luna` |

*Generated by the Merchmix Skills Platform from the current codebase.*