"""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, )