bstore_acumatica_to_databricks

current

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.

main·773cdb33816f·generated by gpt-5.6-luna·9/4/2026, 1:17:55 PM View as Markdown

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

ERP-to-Warehouse Data IngestionproductionexecuteOtherinternalagent facing98%

Automatically transfers retail purchasing, receiving, sales, product, supplier, and warehouse data from the ERP into the analytics warehouse.

The root sync process reads Acumatica streams and loads them into Databricks bronze tables using incremental watermarks, date-chunked extraction, staging tables, and idempotent MERGE operations. Supported streams include PurchaseOrder, PurchaseOrderItems, PurchaseReceipt, SalesInvoice, SalesInvoiceLineItems, StockItem, Vendor, and Warehouse.

Data ReconciliationproductionreadReportinguser facingagent facing97%

Checks whether records in the ERP are present in the Databricks warehouse and identifies missing records.

ReconciliationService compares configured Acumatica streams with Databricks tables, reports source and warehouse counts, identifies missing keys, supports all-stream or selected-stream checks, and can include child records for SalesInvoice and PurchaseOrder.

Data BackfillproductionwriteOtheruser facingagent facing95%

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

The reconciliation and style backfill services fetch missing Acumatica records and can merge them into bronze tables. Requests support date-bounded backfills, key-based backfills, optional child records, and configurable write behavior.

Sales Data AuditingproductionreadReportinguser facingagent facing95%

Traces sales orders, invoices, inventory, and totals through the warehouse layers to find missing or inconsistent sales data.

SalesAuditService audits orders across Bronze, Silver, and Gold tables, supports inventory-based audits, reverse audits, brand and date filters, and Excel-derived order or totals comparisons.

Excel-to-Warehouse Data ComparisonproductionreadReportinguser facingagent facing96%

Compares Acumatica sales or purchase-receipt spreadsheets with warehouse records to show missing and extra documents.

ExcelDiffService and ReceiptDiffService parse uploaded Excel files using calamine with an openpyxl fallback, compare reference or receipt numbers against warehouse keys, apply optional filters and date parsing, and provide prefix-based breakdown metrics.

Invoice Retrieval and InvestigationproductionreadOrdersuser facingagent facing96%

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

InvoiceService supports exact invoice lookups, bounded date-window searches, historical/backdated and changed-invoice queries, line retrieval, and document-user metadata. InvoiceDebugService directly queries both Acumatica SalesInvoice and Invoice contracts for a reference number.

Product and Stock Item LookupproductionreadProductsuser facingagent facing97%

Retrieves product details and current stock held at each warehouse from the ERP.

StockItemService retrieves exact InventoryID records, normalizes Acumatica values and attributes, and summarizes QtyOnHand by warehouse. It accounts for non-stock items and can include or omit zero-quantity warehouse rows.

Style Data Validation and BackfillproductionwriteProductsuser facingagent facing94%

Checks product/style records between Acumatica and Databricks and restores missing product records.

StyleBackfillService validates style records, fetches StockItem data from Acumatica, optionally writes to bronze and silver datasets, reconciles all-time inventory IDs, and supports missing-bronze discovery with pagination and prefix filtering.

Style DeduplicationproductiondestructiveProductsuser facingagent facing96%

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

StyleDedupService previews duplicate groups in bronze style data and atomically materializes a deduplicated silver style table. Apply requires explicit confirmation and supports dry-run SQL validation.

Pipeline Operations MonitoringproductionreadReportinguser facingagent facing94%

Shows whether ingestion streams are fresh and healthy and provides operational run history and lineage information.

OpsService reads stream configuration and warehouse watermarks, reports stream health and summaries, exposes job runs and stream status, caches dashboard views, and provides lineage and filter information.

Pipeline Job OrchestrationproductionexecuteOtheruser facingagent facing92%

Allows authorized operations users or systems to start ingestion, backfill, and stream-specific jobs.

The API can trigger and backfill configured streams through Azure Container Apps in production, with Databricks Jobs as a fallback. Requests support test mode, maximum record limits, date-bounded backfills, and execution identifiers for polling.

Data Validation API AuthenticationproductionreadAuthenticationuser facingagent facing93%

Restricts validation and operational actions to callers that provide the configured API key.

The FastAPI service includes API-key verification utilities and tests that require an X-API-Key header for protected routes. CORS is configured from service settings.

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 handlingdatabricks_dest.py: Databricks SQL staging, table creation, and MERGE loadingsync.py: stream selection and ingestion orchestrationData_Validation/data_validation/main.py: FastAPI application entry pointData_Validation/data_validation/api/v1: versioned HTTP routers for audits, reconciliation, operations, ingestion, invoices, stock items, style backfill, and deduplicationData_Validation/data_validation/services: business logic for audits, reconciliation, backfill, ERP lookups, operations, and deduplicationData_Validation/data_validation/clients: Acumatica, warehouse, Databricks Jobs, and Azure Container Apps clientsData_Validation/data_validation/sql: SQL template loading and renderingData_Validation/data_validation/models: Pydantic request and response schemasdatabricks/resources/jobs.yml and databricks/databricks.yml: Databricks Asset Bundle job deploymentazure-pipelines and bitbucket-pipelines.yml: CI/CD for the sync job and validation API

Patterns

Incremental watermark-based extractionStaging-table then idempotent MERGE loadingLayered API router, service, client, model, and SQL-template architectureRead-only Acumatica access for lookup and audit operationsConfigurable stream definitions loaded from YAMLExplicit confirmation and dry-run protection for destructive table rebuildsShort-lived shared caching for operations dashboard readsSeparate 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.12FastAPIUvicornPydantic v2RequestsDatabricks SQL ConnectorDatabricks SDKpandaspython-calamineopenpyxlPyYAMLSQL templatesDatabricks Delta tablesAzure Container Apps JobsDatabricks Asset BundlesDockerBitbucket PipelinesAzure Pipelinespytestpytest-asynciohttpx

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.