accounting-adapters

Normalized Python interface for public accounting APIs.

accounting-adapters gives you a single, provider-agnostic data model for the core objects every accounting integration has to deal with -- invoices, bills, journal entries, the chart of accounts, customers, and bank transactions -- and a small read adapter for each supported public system that maps that system's wire shape onto the normalized model.

Once two systems speak the same schema, comparing them is one function call (diff_invoices) instead of a pile of per-provider reconciliation code.

Capability matrix

Each adapter implements only the list_* methods its upstream actually supports; anything unsupported raises NotImplementedError rather than returning empty or guessing. The grid below is the source of truth for "which normalized object can I read from which provider."

Provider	Customers	Chart of Accounts	Invoices (AR)	Bills (AP)	Journal Entries	Transactions	Auth model
QuickBooks Online	yes	yes	yes	yes	no	no	OAuth2 bearer + realm_id
Xero	yes	yes	yes	yes	yes	no	OAuth2 bearer + Xero-tenant-id
FreshBooks	yes	no	yes	yes	no	no	OAuth2 bearer + account_id
Wave	yes	no	yes	no	no	no	OAuth2 bearer + business_id (GraphQL)
Bill.com	yes¹	no	no	yes	no	no	devKey + session id
TaxDome	yes²	no	yes	no	no	no	bearer access token
Plaid	no	no	no	no	no	yes	client_id + secret + item token

¹ In Bill.com's accounts-payable context, "customers" are vendors (list_vendors() is an alias for list_customers()). ² TaxDome accounts map onto Customer.

All records share three fields: provider (the source enum), source_id (the upstream id), and an optional raw copy of the original payload for debugging.

Why

Every accounting API has its own object shapes, status vocabularies, pagination scheme, auth model, and rate-limit behavior. If you integrate more than one, you end up writing the same plumbing repeatedly and mapping seven different "invoice" shapes by hand.

This library handles that integration layer for you:

• One normalized model (Pydantic v2) so downstream code reads one schema.
• Per-provider adapters with bearer-token auth plumbing, pagination, and exponential-backoff retry on 429 / 5xx (honoring Retry-After).
• A cross-provider diff over the normalized model, so reconciling two systems is a single call.
• Webhook HMAC verification helpers for each provider, constant-time.
• An offline FakeTransport test harness so the whole thing is testable with no live credentials and no network. See The offline FakeTransport harness.

Scope

This library reads from the named public systems and normalizes the results. It provides a read, normalize, and compare layer for provider data.

Install

pip install accounting-adapters          # from a published wheel
# or, from a checkout:
git clone https://github.com/maxed-oss/accounting-adapters
cd accounting-adapters
pip install -e ".[dev]"

Requires Python 3.9+.

Usage

You supply credentials (the OAuth dance / token refresh is your responsibility). Each adapter exposes streaming list_* iterators and eager all_* collectors.

from accounting_adapters import (
    QuickBooksAdapter, XeroAdapter, FreshBooksAdapter, WaveAdapter, PlaidAdapter,
)

# QuickBooks Online
qbo = QuickBooksAdapter(realm_id="123456789", access_token="ya29...")
for invoice in qbo.list_invoices():
    print(invoice.provider, invoice.number, invoice.status, invoice.total)

# Xero (tenant id + access token)
xero = XeroAdapter(tenant_id="xyz", access_token="...")
accounts = xero.all_accounts()        # List[ChartOfAccounts], normalized types

# FreshBooks (account id + access token)
fb = FreshBooksAdapter(account_id="abc123", access_token="...")
clients = fb.all_customers()

# Wave (GraphQL Business API: business id + access token)
wave = WaveAdapter(business_id="QnVzaW5lc3M6MQ==", access_token="...")
invoices = wave.all_invoices()

# Plaid bank transactions
plaid = PlaidAdapter(
    client_id="...", secret="...", access_token="access-sandbox-...",
    environment="sandbox",
)
for txn in plaid.list_transactions(start_date="2026-01-01"):
    print(txn.posted_at, txn.amount, txn.merchant_name, txn.category)

One call to diff two systems

The headline use case. Read invoices from two different accounting systems and ask where they disagree. The join, the field comparison, and the Decimal/float coercion are handled for you:

from accounting_adapters import diff_invoices, render_report

qbo_invoices = qbo.all_invoices()
xero_invoices = xero.all_invoices()

report = diff_invoices(
    qbo_invoices, xero_invoices, left_label="quickbooks", right_label="xero"
)
print(render_report(report))
# quickbooks vs xero: 12 matched, 1 mismatched, 1 only-quickbooks, 0 only-xero
#   only in quickbooks: 1038
#   ~ 1037: status: DocumentStatus.PAID != DocumentStatus.OPEN; balance: 0.00 != 500.0

if not report.in_sync:
    ...  # surface the drift

diff_invoices and diff_bills join on the document number; diff_customers joins on a case-folded name; diff_accounts joins on the GL code (falling back to name) for reconciling two charts of accounts; diff_transactions joins on the upstream source_id for comparing two bank feeds; and diff_journal_entries joins on the upstream source_id and compares the entry date, memo, currency, and the derived total debits and total credits, so a re-ordered or re-split set of lines that nets to the same totals is correctly reported as matched. The generic diff_records lets you pick any key function and field set. The result (DiffReport) reports left_only, right_only, mismatched (with per-field FieldDeltas), and the matched count, plus an in_sync convenience property and a mismatched_count. If a join key appears more than once on a side, the collision is recorded in duplicate_keys rather than silently distorting the diff (the last record still wins the comparison, as before).

For machine-readable output, DiffReport.to_dict() returns a plain JSON-ready structure (Decimals become strings, enums become their values) so you can emit a diff straight from a CLI or return it over an API with no custom encoder:

import json
print(json.dumps(report.to_dict(), indent=2))
# {
#   "left_label": "quickbooks", "right_label": "xero",
#   "in_sync": false, "matched": 12, "mismatched_count": 1,
#   "left_only": ["1038"], "right_only": [],
#   "mismatched": [{"key": "1037", "deltas": [
#     {"field": "status", "left": "paid", "right": "open"},
#     {"field": "balance", "left": "0.00", "right": "500.0"}]}],
#   "duplicate_keys": {"left": [], "right": []}
# }

A complete, runnable version of this is in examples/diff_two_providers.py -- it runs entirely offline (see below).

Verifying webhooks

from accounting_adapters.webhooks import verify_xero
from accounting_adapters.exceptions import WebhookVerificationError

# In your webhook handler:
try:
    verify_xero(request_body_bytes, request.headers["x-xero-signature"], webhook_key)
except WebhookVerificationError:
    return 401

Helpers exist for QuickBooks, Xero, FreshBooks, Wave, Bill.com, TaxDome, and Plaid. Comparisons are constant-time.

Errors

All adapter errors subclass AdapterError:

• AuthError -- credentials missing/invalid/expired (401/403)
• RateLimitError -- upstream 429 (carries retry_after)
• NotFoundError -- 404
• ApiError -- any other non-2xx (carries status_code)
• WebhookVerificationError -- bad webhook signature
• MappingError -- a payload could not be mapped

The transport retries 429/5xx automatically with exponential backoff + jitter before surfacing these.

The offline FakeTransport harness

A first-class feature of this library is that every adapter is testable offline, with no credentials and no network. Adapters never construct their own HTTP session if you hand them a transport=, so you can inject a queue-driven fake that returns the exact payloads the upstream would return and then assert on the recorded calls.

The harness lives in tests/conftest.py as FakeTransport plus a make_fake pytest fixture:

class FakeTransport(HttpTransport):
    """An HttpTransport whose requests return pre-queued payloads."""

    def __init__(self, responses):
        super().__init__("https://fake.local")
        self._responses = list(responses)
        self.calls = []                       # every request is recorded

    def request(self, method, path, *, params=None, json=None, data=None, headers=None):
        self.calls.append({"method": method, "path": path, "params": params, "json": json})
        return self._responses.pop(0) if self._responses else {}

Using it to test an adapter's mapping and pagination deterministically:

def test_invoice_partial_paid(make_fake):
    adapter = XeroAdapter("tenant-1", "tok", transport=make_fake([fixtures.XERO_INVOICES_PAGE]))
    inv = adapter.all_invoices()[0]
    assert inv.status is DocumentStatus.PARTIALLY_PAID   # mapping
    assert inv.total == Decimal("1100.0")

def test_pagination_offset_advances(make_fake):
    transport = make_fake([fixtures.BILLDOTCOM_BILLS_PAGE])
    BillDotComAdapter("devkey", "session", transport=transport).all_bills()
    assert transport.calls[0]["params"]["start"] == 0   # asserted on recorded call

Why this matters:

• No secrets in CI. The whole suite (and both examples) run with zero credentials, so CI needs no API keys and there is no live-API flakiness.
• Synthetic fixtures only. All fixture payloads in tests/fixtures.py are hand-written and fictional, shaped to mirror each provider's public sandbox docs -- no real client data.
• Deterministic pagination tests. Queue several pages and assert the adapter walks cursors/offsets correctly and stops at the right boundary.

For exercising the real HttpTransport (backoff, Retry-After, the typed error surface) without a fake, the suite uses the responses library to stub HTTP at the requests layer -- see tests/test_transport.py.

Command-line interface

An accounting-adapters CLI lists and diffs normalized records, with a zero-credential --sandbox mode (built-in synthetic data, no network) and --json output for agents:

accounting-adapters providers --json
accounting-adapters list --sandbox --provider quickbooks --resource invoices --json
accounting-adapters diff --sandbox --resource invoices --json   # shows cross-provider drift

Beyond the sandbox, --from-json FILE (for list) and --left-json / --right-json (for diff) operate on files of already-normalized records. The library adds no write path; the CLI only ever reads and compares. --json emits {ok, ...} on success and {ok: false, error: {...}} on a usage error.

Examples

Both run offline against synthetic fixtures (no network, no credentials):

python examples/normalize_invoices.py     # read invoices from QBO + Xero, one schema
python examples/diff_two_providers.py      # ...then diff the two and print the drift

Development & tests

pip install -e ".[dev]"
pytest

The suite is fully offline and uses synthetic fixtures + the FakeTransport harness above -- no real credentials or client data are involved.

Architecture

src/accounting_adapters/
  models.py          # normalized Pydantic models + enums
  transport.py       # HTTP client: auth, pagination, backoff/retry
  diff.py            # cross-provider diff over the normalized model
  webhooks.py        # per-provider HMAC verification helpers
  exceptions.py      # typed error hierarchy
  providers/
    base.py          # BaseAdapter read interface
    quickbooks.py    # QBO Query API       -> normalized
    xero.py          # Xero Accounting API  -> normalized
    freshbooks.py    # FreshBooks API       -> normalized
    wave.py          # Wave GraphQL API     -> normalized
    billdotcom.py    # Bill.com v3 (AP)     -> normalized
    taxdome.py       # TaxDome              -> normalized
    plaid.py         # Plaid transactions   -> normalized

Adding a provider

1. Subclass BaseAdapter, set provider, and implement only the list_* methods the upstream supports (leave the rest to raise NotImplementedError).
2. Write the mapping functions from the provider's payloads to the normalized models.
3. Add synthetic fixtures to tests/fixtures.py and a test that drives the adapter through FakeTransport -- mapping correctness and pagination.
4. Register the new Provider enum value, export the adapter, and add a row to the capability matrix above.

freshbooks.py (REST, money-as-object) and wave.py (GraphQL cursor pagination) are good reference implementations for the two common shapes.

License

Apache-2.0. See LICENSE.