"""QuickBooks Online connector.
Syncs invoices, transactions, customers, and expenses from Intuit's IPP API v3.
Uses QBO queries with pagination (maxResults + startPosition) and upserts
records into dedicated QuickBooks models with qb_doc_id dedup.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ---------------------------------------------------
register_connector(
"quickbooks",
{
"service": "quickbooks",
"display_name": "QuickBooks Online",
"type": "oauth2",
"category": "accounting",
"description": "Sync invoices, transactions, customers, and expenses from QuickBooks Online",
"auth_type": "oauth2",
"auth_fields": ["client_id", "client_secret", "access_token", "refresh_token", "realm_id"],
"capabilities": ["invoices", "transactions", "customers", "expenses"],
"data_types": ["invoices", "transactions", "customers", "expenses"],
"rate_limit": "Varies by plan; typical 150 req/min",
"docs_url": "https://developer.intuit.com/app/developer/qbo/docs/develop",
},
)
# -- Known QBO entity types and their API query keys -------------------------
_QBO_ENTITY_KEYS: Dict[str, str] = {
"Customer": "Customer",
"Invoice": "Invoice",
"Payment": "Payment",
"Expense": "Expense",
"Purchase": "Purchase",
"Bill": "Bill",
"Vendor": "Vendor",
"SalesReceipt": "SalesReceipt",
"CreditMemo": "CreditMemo",
"JournalEntry": "JournalEntry",
"Account": "Account",
}
def _parse_qb_date(date_str: str | None) -> Optional[datetime]:
"""Parse a QuickBooks ISO 8601 date string into a timezone-aware datetime."""
if not date_str:
return None
try:
dt = datetime.fromisoformat(str(date_str).replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except (ValueError, AttributeError):
return None
def _safe_float(value: Any, default: float = 0.0) -> float:
"""Safely convert a value to float."""
if value is None:
return default
try:
return float(value)
except (ValueError, TypeError):
return default
class QuickBooksConnector(OAuthConnector):
"""QuickBooks Online integration — full sync with pagination and upsert."""
_SERVICE = "quickbooks"
_BASE_URL = "https://quickbooks.api.intuit.com/v3/company"
_SANDBOX_BASE_URL = "https://sandbox-quickbooks.api.intuit.com/v3/company"
MAX_RESULTS = 1000
@property
def _base_url(self) -> str:
"""Environment-aware API base URL.
The Intuit OAuth app may be a *sandbox* app — its tokens are only
valid against the sandbox host, and the production host returns
403 ApplicationAuthorizationFailed. Resolution order:
connector config 'environment' > QUICKBOOKS_ENVIRONMENT env/app
config > production default.
"""
import os
from flask import current_app, has_app_context
env = (self.config.get("environment") or "").strip().lower()
if not env and has_app_context():
env = (current_app.config.get("QUICKBOOKS_ENVIRONMENT") or "").strip().lower()
if not env:
env = (os.environ.get("QUICKBOOKS_ENVIRONMENT") or "").strip().lower()
if env == "sandbox":
return self._SANDBOX_BASE_URL
return self._BASE_URL
# -- OAuth 2.0 ------------------------------------------------------------
OAUTH_AUTHORIZE_URL = "https://appcenter.intuit.com/connect/oauth2"
OAUTH_TOKEN_URL = "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer"
OAUTH_SCOPES = [
"com.intuit.quickbooks.accounting",
]
@property
def _oauth_client_id(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_QUICKBOOKS_CLIENT_ID", "")
@property
def _oauth_client_secret(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_QUICKBOOKS_CLIENT_SECRET", "")
@property
def _realm_id(self) -> str:
return self.config.get("realm_id", "")
@property
def _access_token(self) -> str:
return self.config.get("access_token", "")
def oauth_authorize_url(self, state: str) -> str:
"""Build the Intuit OAuth authorization URL."""
from urllib.parse import urlencode
params = {
"client_id": self._oauth_client_id,
"response_type": "code",
"scope": " ".join(self.OAUTH_SCOPES),
"redirect_uri": self._get_redirect_uri(),
"state": state,
"prompt": "consent",
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for access/refresh tokens via Intuit."""
import time as _time
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"QuickBooks OAuth credentials not configured. "
"Set OAUTH_QUICKBOOKS_CLIENT_ID and OAUTH_QUICKBOOKS_CLIENT_SECRET."
)
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self._get_redirect_uri(),
},
auth=(client_id, client_secret),
headers={"Accept": "application/json"},
timeout=15,
)
response.raise_for_status()
data = response.json()
tokens: Dict[str, Any] = {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
# Intuit-specific refresh-token expiry header
x_refresh = response.headers.get("x_refresh_token_expires_in", "")
if x_refresh:
tokens["x_refresh_token_expires_in"] = x_refresh
tokens["refresh_token_expires_at"] = str(int(_time.time()) + int(x_refresh))
return tokens
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh expired access token via Intuit token endpoint."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("QuickBooks OAuth credentials not configured.")
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
},
auth=(client_id, client_secret),
headers={"Accept": "application/json"},
timeout=15,
)
response.raise_for_status()
data = response.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
# -- connect / disconnect -------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate QuickBooks OAuth tokens by querying company info."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
access_token = self._access_token
realm_id = self._realm_id
if not access_token or not realm_id:
raise ValueError("QuickBooks access_token and realm_id are required")
response = self._retry(
_api_get,
f"{self._base_url}/{realm_id}/companyinfo/{realm_id}",
access_token=access_token,
)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
company_info = response.get("CompanyInfo", [{}])[0] if response.get("CompanyInfo") else {}
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"company_name": company_info.get("CompanyName"), "realm_id": realm_id},
)
return {
"status": "connected",
"service": "quickbooks",
"company_name": company_info.get("CompanyName"),
"realm_id": realm_id,
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
"""Revoke QuickBooks access token."""
self._connected = False
try:
access_token = self._access_token
if access_token:
_revoke_token(access_token)
except Exception as exc:
logger.warning("QuickBooks token revocation failed: %s", exc)
self.config.pop("access_token", None)
self.config.pop("refresh_token", None)
self.config.pop("realm_id", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "quickbooks"}
# -- sync -----------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync invoices, transactions, customers, and expenses from QuickBooks."""
from app.models import db
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
if not self._access_token or not self._realm_id:
raise ValueError("QuickBooks access_token and realm_id are required")
# Sync customers first (invoices reference them)
try:
customers = self._sync_customers()
total_records += customers
results["customers"] = customers
db.session.flush()
except Exception as exc:
logger.error("QuickBooks customer sync failed: %s", exc)
results["customers"] = 0
# Sync invoices
try:
invoices = self._sync_invoices()
total_records += invoices
results["invoices"] = invoices
db.session.flush()
except Exception as exc:
logger.error("QuickBooks invoice sync failed: %s", exc)
results["invoices"] = 0
# Sync transactions
try:
transactions = self._sync_transactions()
total_records += transactions
results["transactions"] = transactions
db.session.flush()
except Exception as exc:
logger.error("QuickBooks transaction sync failed: %s", exc)
results["transactions"] = 0
# Sync expenses
try:
expenses = self._sync_expenses()
total_records += expenses
results["expenses"] = expenses
db.session.flush()
except Exception as exc:
logger.error("QuickBooks expense sync failed: %s", exc)
results["expenses"] = 0
db.session.commit()
duration_ms = int((time.monotonic() - start) * 1000)
# If all entities failed to sync (total_records == 0), report error
if total_records == 0:
error_details = ", ".join(
f"{k}=0" for k, v in results.items() if v == 0
)
error_msg = f"QuickBooks sync returned 0 records — possible API credential issue ({error_details})"
self._log(
event_type="sync_error",
status="error",
record_count=0,
duration_ms=duration_ms,
error_message=error_msg,
)
return {
"status": "error",
"error": error_msg,
"record_count": 0,
"duration_ms": duration_ms,
"details": results,
}
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
db.session.rollback()
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- Entity sync methods --------------------------------------------------
def _sync_customers(self) -> int:
"""Fetch and upsert QuickBooks customers."""
from app.models import db, QuickbooksCustomer
items = self._qbo_query_all("SELECT * FROM Customer")
count = 0
for raw in items:
qb_id = raw.get("Id", "")
if not qb_id:
continue
customer = QuickbooksCustomer.query.filter_by(
company_id=self.company_id,
qb_doc_id=qb_id,
).first()
if not customer:
customer = QuickbooksCustomer(
company_id=self.company_id,
qb_doc_id=qb_id,
qb_realm_id=self._realm_id,
)
db.session.add(customer)
db.session.flush()
customer.display_name = raw.get("Name", customer.display_name) or customer.display_name
customer.email = (raw.get("PrimaryEmailAddr", {}).get("Address", customer.email) or customer.email)
customer.phone = (raw.get("PrimaryPhone", {}).get("FreeFormNumber", customer.phone) or customer.phone)
customer.balance = _safe_float(raw.get("Balance"), customer.balance)
# Parse billing address
bill_addr = raw.get("BillAddr", {})
if bill_addr:
customer.billing_address_json = {
"line1": bill_addr.get("Line1", ""),
"line2": bill_addr.get("Line2", ""),
"city": bill_addr.get("City", ""),
"state": bill_addr.get("CountrySubDivisionCode", bill_addr.get("State", "")),
"postal_code": bill_addr.get("PostalCode", ""),
"country": bill_addr.get("Country", ""),
}
customer.metadata_json = raw
count += 1
logger.info("QuickBooks customers synced: %d records", count)
return count
def _sync_invoices(self) -> int:
"""Fetch and upsert QuickBooks invoices."""
from app.models import db, QuickbooksInvoice
items = self._qbo_query_all("SELECT * FROM Invoice")
count = 0
for raw in items:
qb_id = raw.get("Id", "")
if not qb_id:
continue
invoice = QuickbooksInvoice.query.filter_by(
qb_doc_id=qb_id,
).first()
if not invoice:
invoice = QuickbooksInvoice(
qb_doc_id=qb_id,
company_id=self.company_id,
qb_realm_id=self._realm_id,
)
db.session.add(invoice)
db.session.flush()
invoice.customer_name = (
raw.get("CustomerRef", {}).get("name", invoice.customer_name) or invoice.customer_name
)
invoice.invoice_num = raw.get("DocNumber", invoice.invoice_num) or invoice.invoice_num
invoice.total_amount = _safe_float(raw.get("TotalAmt"), invoice.total_amount)
invoice.tax_amount = _safe_float(raw.get("TxnTaxAmount"), invoice.tax_amount)
invoice.status = raw.get("Status", invoice.status) or invoice.status
invoice.due_date = _parse_qb_date(raw.get("DueDate")) or invoice.due_date
invoice.tx_date = _parse_qb_date(raw.get("TxnDate")) or invoice.tx_date
# Parse line items
line_items = raw.get("Line", [])
invoice.line_items_json = [
{
"description": line.get("Description", ""),
"amount": _safe_float(line.get("Amount")),
"qty": _safe_float(line.get("Qty")),
"unit_price": _safe_float(line.get("UnitPrice")),
"account_name": line.get("DetailType", "AccountBasedExpenseLineDetail") == "AccountBasedExpenseLineDetail"
and line.get("AccountBasedExpenseLineDetail", {}).get("AccountName", ""),
}
for line in line_items if isinstance(line, dict)
]
invoice.metadata_json = raw
count += 1
logger.info("QuickBooks invoices synced: %d records", count)
return count
def _sync_transactions(self) -> int:
"""Fetch and upsert QuickBooks transactions."""
from app.models import db, QuickbooksTransaction
# Sync various transaction types
entity_types = [
("JournalEntry", "SELECT * FROM JournalEntry"),
("Payment", "SELECT * FROM Payment"),
("CreditMemo", "SELECT * FROM CreditMemo"),
("Bill", "SELECT * FROM Bill"),
("SalesReceipt", "SELECT * FROM SalesReceipt"),
]
count = 0
for entity_type, query in entity_types:
items = self._qbo_query_all(query)
for raw in items:
qb_id = raw.get("Id", "")
if not qb_id:
continue
tx = QuickbooksTransaction.query.filter_by(
qb_doc_id=qb_id,
).first()
if not tx:
tx = QuickbooksTransaction(
qb_doc_id=qb_id,
company_id=self.company_id,
qb_realm_id=self._realm_id,
tx_type=entity_type,
)
db.session.add(tx)
db.session.flush()
tx.tx_type = entity_type
tx.tx_date = _parse_qb_date(raw.get("TxnDate")) or tx.tx_date
tx.amount = _safe_float(raw.get("TotalAmt", raw.get("Amount")), tx.amount)
tx.account_name = (
raw.get("AccountRef", {}).get("name", tx.account_name) or tx.account_name
)
tx.account_id = raw.get("AccountRef", {}).get("value", tx.account_id) or tx.account_id
tx.memo = raw.get("Memo", tx.memo) or tx.memo
tx.class_name = raw.get("ClassRef", {}).get("name", tx.class_name) or tx.class_name
tx.metadata_json = raw
count += 1
logger.info("QuickBooks transactions synced: %d records", count)
return count
def _sync_expenses(self) -> int:
"""Fetch and upsert QuickBooks expenses."""
from app.models import db, QuickbooksExpense
items = self._qbo_query_all("SELECT * FROM Purchase")
count = 0
for raw in items:
qb_id = raw.get("Id", "")
if not qb_id:
continue
expense = QuickbooksExpense.query.filter_by(
company_id=self.company_id,
qb_doc_id=qb_id,
).first()
if not expense:
expense = QuickbooksExpense(
company_id=self.company_id,
qb_doc_id=qb_id,
qb_realm_id=self._realm_id,
)
db.session.add(expense)
db.session.flush()
expense.account_name = (
raw.get("AccountRef", {}).get("name", expense.account_name) or expense.account_name
)
expense.amount = _safe_float(raw.get("Amount"), expense.amount)
expense.vendor_name = (
raw.get("VendorRef", {}).get("name", expense.vendor_name) or expense.vendor_name
)
expense.tx_date = _parse_qb_date(raw.get("TxnDate")) or expense.tx_date
expense.memo = raw.get("Memo", expense.memo) or expense.memo
expense.category = raw.get("ClassRef", {}).get("name", expense.category) or expense.category
expense.metadata_json = raw
count += 1
logger.info("QuickBooks expenses synced: %d records", count)
return count
# -- Paginated QBO query --------------------------------------------------
def _qbo_query_all(self, query: str) -> List[Dict[str, Any]]:
"""Execute a QBO query with pagination and return all results."""
access_token = self._access_token
realm_id = self._realm_id
if not access_token or not realm_id:
raise ValueError("QuickBooks access_token and realm_id are required")
all_items: List[Dict[str, Any]] = []
start_position = 1
max_results = self.MAX_RESULTS
while True:
self._check_rate_limit()
response = self._retry(
_api_query,
f"{self._base_url}/{realm_id}/query",
access_token=access_token,
query=query,
start_position=start_position,
max_results=max_results,
)
query_response = response.get("QueryResponse", {})
items = _extract_items(query_response, query)
if not items:
break
all_items.extend(items)
logger.debug(
"QuickBooks %s: fetched %d (total %d)",
query, len(items), len(all_items),
)
# Check if we got fewer items than requested (last page)
if len(items) < max_results:
break
start_position += len(items)
# Small pause between pages to stay within rate limits
time.sleep(0.15)
logger.info("QuickBooks: total %d records for %s", len(all_items), query.split("FROM")[1].strip() if "FROM" in query else query)
return all_items
# -- status ---------------------------------------------------------------
def status(self) -> Dict[str, Any]:
from app.models import (
QuickbooksInvoice, QuickbooksTransaction,
QuickbooksCustomer, QuickbooksExpense,
)
if not self._connected and not self._access_token:
return {
"service": "quickbooks",
"connected": False,
"config_present": bool(self._access_token) and bool(self._realm_id),
}
try:
response = self._retry(
_api_get,
f"{self._base_url}/{self._realm_id}/companyinfo/{self._realm_id}",
access_token=self._access_token,
)
company_info = response.get("CompanyInfo", [{}])[0] if response.get("CompanyInfo") else {}
# Record counts
invoice_count = QuickbooksInvoice.query.filter_by(company_id=self.company_id).count()
transaction_count = QuickbooksTransaction.query.filter_by(company_id=self.company_id).count()
customer_count = QuickbooksCustomer.query.filter_by(company_id=self.company_id).count()
expense_count = QuickbooksExpense.query.filter_by(company_id=self.company_id).count()
return {
"service": "quickbooks",
"connected": True,
"company_name": company_info.get("CompanyName"),
"realm_id": self._realm_id,
"last_sync_at": self.config.get("last_sync_at"),
"record_counts": {
"invoices": invoice_count,
"transactions": transaction_count,
"customers": customer_count,
"expenses": expense_count,
},
}
except Exception as exc:
return {"service": "quickbooks", "connected": False, "error": str(exc)}
# -- Register in the framework registry ---------------------------------------
_REGISTRY["quickbooks"] = QuickBooksConnector
# -- Internal helpers ---------------------------------------------------------
def _extract_items(query_response: Dict[str, Any], query: str) -> List[Dict[str, Any]]:
"""Extract the entity list from a QBO QueryResponse payload."""
entity_name = ""
from_part = query.split("FROM")
if len(from_part) > 1:
entity_name = from_part[1].strip().split()[0]
# Try explicit mapping
mapped_key = _QBO_ENTITY_KEYS.get(entity_name)
if mapped_key and mapped_key in query_response:
items = query_response[mapped_key]
if isinstance(items, list):
return items
# Case-insensitive fallback
entity_lower = entity_name.lower()
for key, value in query_response.items():
if key.lower() == entity_lower and isinstance(value, list):
return value
# Generic fallback: first key that isn't pagination metadata
skip_keys = {"startPosition", "maxResults", "totalCount", "SyncToken"}
for key, value in query_response.items():
if key not in skip_keys and isinstance(value, list):
return value
return []
def _api_get(url: str, *, access_token: str) -> Dict[str, Any]:
"""Make a GET request to the Intuit IPP API."""
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
"Content-Type": "application/json",
}
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
return resp.json()
def _api_query(
url: str,
*,
access_token: str,
query: str,
start_position: int = 1,
max_results: int = 1000,
) -> Dict[str, Any]:
"""Execute a QBO query.
QBO's /query endpoint does NOT accept a JSON payload — the query is
passed as a URL parameter (GET) and pagination is embedded in the
query string itself via STARTPOSITION / MAXRESULTS.
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
"Accept-Encoding": "gzip",
}
paged_query = f"{query} STARTPOSITION {start_position} MAXRESULTS {max_results}"
resp = requests.get(
url,
headers=headers,
params={"query": paged_query, "minorversion": "75"},
timeout=30,
)
resp.raise_for_status()
return resp.json()
def _revoke_token(access_token: str) -> None:
"""Revoke a QuickBooks OAuth token."""
requests.post(
"https://developer.api.intuit.com/v1/oauth2/tokens/revoke",
data={"token": access_token},
timeout=15,
)