"""Connector management API routes.
Endpoints for the SPA frontend to discover, create, configure, and manage
external integration connectors (HubSpot, QuickBooks, Google Ads, …).
"""
from __future__ import annotations
import logging
import time
from flask import Blueprint, request, jsonify, g
from flask_login import current_user
from datetime import datetime, timezone
from ..models import db, Connector, ConnectorLog, UserCompany, Company
from .api_proxy import require_auth_json
from app.utils.csrf import require_csrf
from app import limiter
logger = logging.getLogger(__name__)
# -- Tier configuration -----------------------------------------------------
# Limits are the canonical per-tier limits from app.utils.feature_gate
# (starter=1, launch=3, growth=10, command=50).
from app.utils.feature_gate import get_company_tier, get_tier_limits
# get_company_tier() normalizes 'enterprise' → 'command'
TIER_REAL_TIME_ALLOWED = {"command", "enterprise"}
def _is_super_admin() -> bool:
"""True when the current user is a super admin (bypasses all tier gates)."""
return getattr(current_user, "role", "") == "super_admin"
def _get_tier(company_id: str) -> str:
"""Return the company's effective subscription tier (via feature_gate)."""
company = db.session.get(Company, company_id)
return get_company_tier(company)
def _get_tier_info(company_id: str) -> dict:
"""Return tier, max connectors, and current count for the company."""
tier = _get_tier(company_id)
max_connectors = get_tier_limits(tier)["connectors"]
current_count = Connector.query.filter_by(company_id=company_id).count()
return {
"tier": tier,
"connector_limit": max_connectors,
"connectors_used": current_count,
}
connectors_bp = Blueprint("connectors", __name__)
# -- helpers ----------------------------------------------------------------
def _get_company_id() -> str | None:
"""Return the company ID for the current request.
Checks g.proxy_company_id first (set by super-admin proxy routes),
then falls back to the user's own company.
"""
if hasattr(g, 'proxy_company_id') and g.proxy_company_id:
return g.proxy_company_id
user_company = UserCompany.query.filter_by(user_id=current_user.id).first()
return user_company.company_id if user_company else None
def _json_error(msg: str, code: int) -> tuple:
return jsonify({"error": msg}), code
# -- Available integrations -------------------------------------------------
@connectors_bp.route("/api/connectors/available", methods=["GET"])
@require_auth_json()
def list_available():
"""Return the catalogue of integrations the platform supports."""
from app.connectors import get_available_connectors
connectors = get_available_connectors()
# Map display_name → name for frontend compatibility
for c in connectors:
if "display_name" in c and "name" not in c:
c["name"] = c["display_name"]
return jsonify({"connectors": connectors})
# -- CRUD -------------------------------------------------------------------
@connectors_bp.route("/api/connectors", methods=["GET"])
@require_auth_json()
def list_connectors():
"""List all connectors for the current company."""
company_id = _get_company_id()
if not company_id:
return jsonify({
"connectors": [],
"message": "No company found. Set up your company to enable integrations."
})
items = Connector.query.filter_by(company_id=company_id).order_by(
Connector.created_at.desc()
).all()
from app.connectors import get_connector_class, OAuthConnector, get_connector_metadata
tier_info = _get_tier_info(company_id)
return jsonify({
"connectors": [
{
"id": c.id,
"service": c.service,
"serviceDisplayName": (get_connector_metadata(c.service) or {}).get("display_name", c.service.replace("_", " ").title()),
"status": c.status,
"auth_type": "oauth2" if issubclass(get_connector_class(c.service) or object, OAuthConnector) else "manual",
"sync_frequency": c.sync_frequency,
"last_sync_at": c.last_sync_at.isoformat() if c.last_sync_at else None,
"error_message": c.error_message,
"created_at": c.created_at.isoformat() if c.created_at else None,
}
for c in items
],
**tier_info,
})
@connectors_bp.route("/api/connectors", methods=["POST"])
@require_auth_json()
@require_csrf
def create_connector():
"""Create a new connector configuration.
Body:
service (str) — e.g. 'hubspot', 'quickbooks', 'google_ads'
config (obj) — service-specific credentials
sync_frequency (str, optional) — 'real_time' | 'hourly' | 'daily'
"""
data = request.get_json(silent=True) or {}
service = data.get("service", "").strip().lower()
config = data.get("config", {})
sync_frequency = data.get("sync_frequency", "hourly")
company_id = _get_company_id()
if not company_id:
return _json_error("No company found. Set up your company first.", 400)
if not service:
return _json_error("service is required", 400)
# Validate that the service is registered
from app.connectors import get_connector_class
if get_connector_class(service) is None:
return _json_error(f"Unknown service: {service}", 400)
# Validate sync_frequency
if sync_frequency not in ("real_time", "hourly", "daily"):
return _json_error("sync_frequency must be one of: real_time, hourly, daily", 400)
# --- Tier enforcement ---
# Super admins bypass all tier gates.
if not _is_super_admin():
# Gate real-time sync by tier
if sync_frequency == "real_time" and _get_tier(company_id) not in TIER_REAL_TIME_ALLOWED:
return _json_error("Real-time sync requires Command or Enterprise plan.", 403)
# Check connector count limit
tier_info = _get_tier_info(company_id)
if tier_info["connectors_used"] >= tier_info["connector_limit"]:
return _json_error(
f"Connector limit reached ({tier_info['connectors_used']}/{tier_info['connector_limit']}). "
"Upgrade your plan to add more.",
403,
)
# Check for duplicate active connector for this service
existing = Connector.query.filter_by(company_id=company_id, service=service).first()
if existing and existing.status in ("connected", "syncing"):
return _json_error(f"Active {service} connector already exists", 409)
# Try to connect immediately if credentials are provided
connector_obj = None
status = "inactive"
error_message = ""
if config:
try:
from app.connectors import build_connector
connector_obj = build_connector(
service=service,
company_id=company_id,
config=config,
)
result = connector_obj.connect()
if result.get("status") == "connected":
status = "connected"
else:
error_message = result.get("error", "Connection failed")
status = "error"
except Exception as exc:
error_message = str(exc)
status = "error"
# Persist the connector record
connector_record = Connector(
company_id=company_id,
service=service,
status=status,
sync_frequency=sync_frequency,
error_message=error_message,
)
# Use the .config property to encrypt credentials at rest
connector_record.config = config
db.session.add(connector_record)
db.session.commit()
# If we created a connector object, link it to the DB record
if connector_obj:
connector_obj.connector_id = connector_record.id
connector_obj.config["last_sync_at"] = connector_record.last_sync_at
return jsonify({
"success": True,
"connector": {
"id": connector_record.id,
"service": connector_record.service,
"status": connector_record.status,
"sync_frequency": connector_record.sync_frequency,
"error_message": connector_record.error_message,
}
}), 201
@connectors_bp.route("/api/connectors/<connector_id>", methods=["DELETE"])
@require_auth_json()
@require_csrf
def delete_connector(connector_id: str):
"""Disconnect and remove a connector."""
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
# Attempt a clean disconnect (best effort)
try:
from app.connectors import build_connector
conn = build_connector(connector.service, company_id, connector.config, connector.id)
conn.disconnect()
except Exception:
pass
# Delete associated logs first (FK constraint prevents cascade on connector_id)
ConnectorLog.query.filter_by(connector_id=connector.id).delete()
db.session.delete(connector)
db.session.commit()
return jsonify({"success": True, "message": f"{connector.service} connector removed"})
@connectors_bp.route("/api/connectors/<connector_id>", methods=["PUT"])
@require_auth_json()
@require_csrf
def update_connector(connector_id: str):
"""Update a connector configuration.
Body (all optional):
sync_frequency (str) — 'real_time' | 'hourly' | 'daily'
config (obj) — service-specific credentials
"""
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
data = request.get_json(silent=True) or {}
# --- Sync frequency update with tier gate ---
sync_frequency = data.get("sync_frequency")
if sync_frequency is not None:
if sync_frequency not in ("real_time", "hourly", "daily"):
return _json_error("sync_frequency must be one of: real_time, hourly, daily", 400)
if (
sync_frequency == "real_time"
and not _is_super_admin()
and _get_tier(company_id) not in TIER_REAL_TIME_ALLOWED
):
return _json_error("Real-time sync requires Command or Enterprise plan.", 403)
connector.sync_frequency = sync_frequency
# --- Config update ---
config = data.get("config")
if config is not None:
connector.config = config
db.session.commit()
return jsonify({
"success": True,
"connector": {
"id": connector.id,
"service": connector.service,
"status": connector.status,
"sync_frequency": connector.sync_frequency,
}
})
# -- Actions ----------------------------------------------------------------
@connectors_bp.route("/api/connectors/<connector_id>/status", methods=["GET"])
@require_auth_json()
def connector_status(connector_id: str):
"""Check the live connection status of a connector."""
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
try:
from app.connectors import build_connector
conn = build_connector(connector.service, company_id, connector.config, connector.id)
result = conn.status()
return jsonify({
"connector_id": connector.id,
"service": connector.service,
"db_status": connector.status,
"live_status": result,
})
except Exception as exc:
return jsonify({
"connector_id": connector.id,
"service": connector.service,
"db_status": connector.status,
"live_status": {"connected": False, "error": str(exc)},
})
@connectors_bp.route("/api/connectors/<connector_id>/sync", methods=["POST"])
@require_auth_json()
@require_csrf
def trigger_sync(connector_id: str):
"""Manually trigger a data sync for the connector."""
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
if connector.status == "syncing":
return _json_error("Sync already in progress", 409)
connector.status = "syncing"
connector.error_message = ""
db.session.commit()
try:
from app.connectors import build_connector
conn = build_connector(connector.service, company_id, connector.config, connector.id)
result = conn.sync()
connector.status = "connected" if result.get("status") == "success" else "error"
connector.last_sync_at = datetime.now(timezone.utc)
connector.error_message = result.get("error", "")
db.session.commit()
return jsonify({
"success": True,
"status": result.get("status"),
"record_count": result.get("record_count", 0),
"duration_ms": result.get("duration_ms"),
"details": result.get("details"),
"last_sync_at": connector.last_sync_at.isoformat(),
})
except Exception as exc:
connector.status = "error"
connector.error_message = str(exc)
db.session.commit()
return _json_error(str(exc), 500)
@connectors_bp.route("/api/connectors/<connector_id>/logs", methods=["GET"])
@require_auth_json()
def connector_logs(connector_id: str):
"""Return sync logs for a specific connector.
Query params:
limit (int, default 50) — max entries to return
offset (int, default 0) — skip N entries
"""
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
limit = request.args.get("limit", 50, type=int)
offset = request.args.get("offset", 0, type=int)
logs = ConnectorLog.query.filter_by(
company_id=company_id,
connector_id=connector_id,
).order_by(ConnectorLog.created_at.desc()).offset(offset).limit(limit).all()
return jsonify({
"connector_id": connector_id,
"logs": [
{
"id": l.id,
"event_type": l.event_type,
"status": l.status,
"record_count": l.record_count,
"duration_ms": l.duration_ms,
"error_message": l.error_message,
"details": l.details_json,
"created_at": l.created_at.isoformat() if l.created_at else None,
}
for l in logs
],
"total": len(logs),
})
# -- OAuth 2.0 endpoints ----------------------------------------------------
@connectors_bp.route("/api/connectors/oauth/config/<service>", methods=["GET"])
@require_auth_json()
def oauth_config(service: str):
"""Return OAuth config metadata for a service.
Tells the frontend whether this service uses OAuth and whether a
redirect is needed (vs showing a manual config modal).
"""
from app.connectors import get_connector_class, OAuthConnector
cls = get_connector_class(service)
if cls is None:
return _json_error(f"Unknown service: {service}", 404)
uses_oauth = issubclass(cls, OAuthConnector) if isinstance(cls, type) else False
return jsonify({
"service": service,
"uses_oauth": uses_oauth,
"auth_type": "oauth2" if uses_oauth else "manual",
"redirect_needed": uses_oauth,
})
@connectors_bp.route("/api/connectors/oauth/authorize/<service>", methods=["GET"])
@require_auth_json()
@limiter.limit("10 per minute")
def oauth_authorize(service: str):
"""Start the OAuth 2.0 authorization flow.
Generates a CSRF state token, stores it server-side, and redirects
the user to the provider's authorization page.
"""
from app.connectors import get_connector_class, OAuthConnector, build_connector, generate_oauth_state
cls = get_connector_class(service)
if cls is None:
return _json_error(f"Unknown service: {service}", 404)
if not issubclass(cls, OAuthConnector):
return _json_error(f"{service} does not support OAuth", 400)
company_id = _get_company_id()
if not company_id:
return _json_error("No company found. Set up your company first.", 400)
# Check tier limits (super admins bypass)
if not _is_super_admin():
tier_info = _get_tier_info(company_id)
if tier_info["connectors_used"] >= tier_info["connector_limit"]:
return _json_error(
f"Connector limit reached ({tier_info['connectors_used']}/{tier_info['connector_limit']}). "
"Upgrade your plan to add more.",
403,
)
# Check for existing connector that needs to be re-authenticated
existing = Connector.query.filter_by(company_id=company_id, service=service).first()
connector_id = existing.id if existing else ""
# Allow caller to specify the redirect path after OAuth (e.g., /admin/integrations)
next_path = request.args.get("next", "")
# Generate CSRF state
state = generate_oauth_state(service, company_id, connector_id, next_path=next_path)
# Build a temporary connector to get the authorize URL
conn = cls(company_id=company_id, config={}, connector_id=connector_id)
redirect_uri = conn._get_redirect_uri()
try:
authorize_url = conn.oauth_authorize_url(state)
except Exception as exc:
logger.error("Failed to build OAuth authorize URL for %s: %s", service, exc)
return _json_error(f"OAuth configuration error for {service}", 500)
# Append redirect_uri if the provider URL doesn't already have it
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
parsed = urlparse(authorize_url)
params = parse_qs(parsed.query)
if "redirect_uri" not in params:
params["redirect_uri"] = redirect_uri
parsed = parsed._replace(query=urlencode(params, doseq=True))
authorize_url = urlunparse(parsed)
from flask import redirect
return redirect(authorize_url, code=302)
@connectors_bp.route("/api/connectors/oauth/callback/<service>", methods=["GET"])
@limiter.limit("10 per minute")
def oauth_callback(service: str):
"""Handle the OAuth 2.0 authorization callback.
Expects ``code`` and ``state`` query parameters from the provider.
Exchanges the code for tokens, then creates/updates the connector record.
"""
from app.connectors import get_connector_class, OAuthConnector, build_connector, verify_oauth_state
cls = get_connector_class(service)
if cls is None:
return _json_error(f"Unknown service: {service}", 404)
if not issubclass(cls, OAuthConnector):
return _json_error(f"{service} does not support OAuth", 400)
code = request.args.get("code", "")
state = request.args.get("state", "")
error = request.args.get("error", "")
error_description = request.args.get("error_description", "")
# QuickBooks returns realmId in callback params (company ID needed for API calls)
realm_id = request.args.get("realmId", "")
from flask import redirect, url_for
base_url = url_for("dashboard.index", _external=True).rstrip("/")
integrations_url = base_url.replace("/dashboard", "") + "/admin/integrations"
if error:
msg = error_description or error
return redirect(f"{integrations_url}?oauth_error={service}&message={msg}"), 302
if not code or not state:
return redirect(f"{integrations_url}?oauth_error={service}&message=Missing code or state parameter"), 302
# Verify CSRF state
try:
state_data = verify_oauth_state(service, state)
except ValueError as exc:
return redirect(f"{integrations_url}?oauth_error={service}&message={exc}"), 302
company_id = state_data["company_id"]
connector_id = state_data.get("connector_id", "")
# Redirect to /admin/integrations which is served by admin_spa() → SPA shell.
redirect_base = integrations_url
# Check tier limits (super admins bypass)
if not _is_super_admin():
tier_info = _get_tier_info(company_id)
if not connector_id and tier_info["connectors_used"] >= tier_info["connector_limit"]:
return redirect(f"{redirect_base}?oauth_error={service}&message=Connector limit reached"), 302
# Exchange code for tokens
try:
conn = cls(company_id=company_id, config={}, connector_id=connector_id)
# Pass PKCE verifier if available (e.g., Google Sheets uses PKCE)
pkce_verifier = state_data.get("pkce_verifier", "")
tokens = conn.exchange_code_for_tokens(code, pkce_verifier=pkce_verifier)
except Exception as exc:
logger.error("OAuth token exchange failed for %s: %s", service, exc)
return redirect(f"{redirect_base}?oauth_error={service}&message=Token exchange failed: {exc}"), 302
# Build config from tokens — track when tokens were obtained
config = dict(tokens)
config["token_obtained_at"] = str(int(time.time()))
# Add realm_id from callback params (QuickBooks-specific)
if realm_id:
config["realm_id"] = realm_id
# Try to connect and get service-specific metadata (e.g., realm_id for QB)
# NOTE: QuickBooks does NOT return realm_id in the token response.
# The user must provide realm_id when configuring the connector.
try:
conn = cls(company_id=company_id, config=config, connector_id=connector_id)
result = conn.connect()
if result.get("status") == "connected":
# Merge any service-specific metadata back into config
if result.get("realm_id"):
config["realm_id"] = result["realm_id"]
else:
error_msg = result.get("error", "Connection validation failed")
logger.warning("OAuth connect validation failed for %s: %s", service, error_msg)
# Still save the connector so the user can troubleshoot
except Exception as exc:
logger.warning("OAuth connect validation warning for %s: %s", service, exc)
# Create or update connector record
if connector_id:
# Update existing connector
connector = Connector.query.filter_by(id=connector_id, company_id=company_id).first()
if connector:
connector.config = config
connector.status = "connected"
connector.error_message = ""
else:
connector = Connector(
company_id=company_id,
service=service,
status="connected",
sync_frequency="hourly",
)
connector.config = config
db.session.add(connector)
else:
# Create new connector
connector = Connector(
company_id=company_id,
service=service,
status="connected",
sync_frequency="hourly",
)
connector.config = config
db.session.add(connector)
db.session.commit()
# For google_sheets, redirect to spreadsheet picker so user can select a sheet
if service == "google_sheets":
return redirect(f"{redirect_base}?oauth_success={service}&picker=google_sheets"), 302
return redirect(f"{redirect_base}?oauth_success={service}"), 302
@connectors_bp.route("/api/connectors/oauth/connected-sheets", methods=["GET"])
@require_auth_json()
def oauth_connected_sheets():
"""List user's spreadsheets for the spreadsheet picker.
Requires an active google_sheets connector with valid OAuth tokens.
Returns a list of spreadsheets via Drive API so the user can pick one.
"""
from app.connectors.google_sheets import _build_oauth_credentials, _list_spreadsheets
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
company_id=company_id, service="google_sheets"
).first()
if not connector:
return _json_error("No Google Sheets connector found. Please connect first.", 404)
config = connector.config
access_token = config.get("access_token", "")
if not access_token:
return _json_error("No access token available. Please reconnect.", 401)
try:
credentials = _build_oauth_credentials(config)
spreadsheets = _list_spreadsheets(credentials)
return jsonify({
"success": True,
"spreadsheets": spreadsheets,
"connector_id": connector.id,
})
except Exception as exc:
logger.error("Failed to list spreadsheets: %s", exc)
return _json_error(f"Failed to list spreadsheets: {exc}", 500)
@connectors_bp.route("/api/connectors/oauth/connected-sheets/select", methods=["POST"])
@require_auth_json()
@require_csrf
def oauth_select_spreadsheet():
"""Select a spreadsheet for the google_sheets connector.
Body:
spreadsheet_id (str) — Google Sheets spreadsheet ID
Updates the connector config with the selected spreadsheet and triggers
an initial sync.
"""
from app.connectors import build_connector
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
data = request.get_json(silent=True) or {}
spreadsheet_id = data.get("spreadsheet_id", "").strip()
if not spreadsheet_id:
return _json_error("spreadsheet_id is required", 400)
connector = Connector.query.filter_by(
company_id=company_id, service="google_sheets"
).first()
if not connector:
return _json_error("No Google Sheets connector found. Please connect first.", 404)
# Update config with selected spreadsheet
updated_config = dict(connector.config)
updated_config["spreadsheet_id"] = spreadsheet_id
connector.config = updated_config
connector.status = "connected"
db.session.commit()
# Trigger initial sync
try:
conn = build_connector(
"google_sheets", company_id, connector.config, connector.id
)
result = conn.sync()
connector.status = "connected" if result.get("status") == "success" else "error"
connector.last_sync_at = datetime.now(timezone.utc)
connector.error_message = result.get("error", "")
db.session.commit()
return jsonify({
"success": True,
"message": f"Spreadsheet selected: {spreadsheet_id}",
"sync_result": result,
})
except Exception as exc:
logger.error("Initial sync failed after spreadsheet selection: %s", exc)
return _json_error(f"Spreadsheet selected but initial sync failed: {exc}", 500)
@connectors_bp.route("/api/connectors/<connector_id>/refresh", methods=["POST"])
@require_auth_json()
@limiter.limit("5 per minute")
@require_csrf
def refresh_connector(connector_id: str):
"""Refresh an OAuth access token for a connector.
For OAuth connectors: calls the provider's token refresh endpoint
and updates the stored tokens.
For non-OAuth connectors: returns a message that refresh is not applicable.
"""
from app.connectors import get_connector_class, OAuthConnector, build_connector
company_id = _get_company_id()
if not company_id:
return _json_error("No company found.", 400)
connector = Connector.query.filter_by(
id=connector_id, company_id=company_id
).first()
if not connector:
return _json_error("Connector not found", 404)
cls = get_connector_class(connector.service)
if cls is None:
return _json_error(f"Unknown service: {connector.service}", 400)
if not issubclass(cls, OAuthConnector):
return _json_error(f"{connector.service} does not support token refresh", 400)
refresh_token = connector.config.get("refresh_token", "")
if not refresh_token:
return _json_error("No refresh token available for this connector", 400)
try:
conn = build_connector(
connector.service, company_id, connector.config, connector.id
)
new_tokens = conn.refresh_access_token(refresh_token)
# Update stored tokens
updated_config = dict(connector.config)
updated_config.update(new_tokens)
updated_config["token_obtained_at"] = str(int(time.time()))
connector.config = updated_config
# Update token expiry tracking
expires_in = new_tokens.get("expires_in")
if expires_in:
from datetime import timedelta
connector.last_sync_at = datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))
db.session.commit()
return jsonify({
"success": True,
"message": "Token refreshed successfully",
"expires_in": expires_in,
})
except Exception as exc:
logger.error("Token refresh failed for %s: %s", connector.service, exc)
return _json_error(f"Token refresh failed: {exc}", 500)
# -- Webhook endpoints ------------------------------------------------------
@connectors_bp.route("/api/connectors/webhooks/<service>", methods=["POST"])
@limiter.limit("60 per minute")
def webhook_receiver(service: str):
"""
Generic webhook receiver for connectors that use webhook delivery.
Currently supports:
- angi: Angi lead feed webhook
The service parameter identifies which connector should process the webhook.
Args:
service: Service identifier (e.g., 'angi')
Returns:
JSON response indicating success or failure
"""
import logging
logger = logging.getLogger(__name__)
# Get the webhook payload
payload = request.get_json(silent=True)
if not payload:
logger.warning(f"Empty or invalid JSON payload for {service} webhook")
return jsonify({
"success": False,
"error": "Invalid JSON payload"
}), 400
try:
# Find the connector for this service
connector = Connector.query.filter_by(
service=service,
status="connected"
).first()
if not connector:
logger.warning(f"No active connector found for service: {service}")
return jsonify({
"success": False,
"error": f"No active {service} connector configured"
}), 404
# Build the connector instance
from app.connectors import build_connector
conn = build_connector(
service=service,
company_id=connector.company_id,
config=connector.config,
connector_id=connector.id
)
# Check if this connector has a process_webhook method
if not hasattr(conn, 'process_webhook'):
logger.error(f"Connector {service} does not support webhooks")
return jsonify({
"success": False,
"error": f"Webhook not supported by {service}"
}), 501
# Process the webhook
result = conn.process_webhook(
payload=payload,
request_headers=dict(request.headers),
)
if result.get('success'):
logger.info(f"{service} webhook processed: {result.get('lead_id', 'unknown')}")
return jsonify(result), 200
else:
logger.error(f"{service} webhook processing failed: {result.get('error')}")
return jsonify(result), 400
except Exception as e:
logger.error(f"Webhook processing error for {service}: {str(e)}", exc_info=True)
return jsonify({
"success": False,
"error": "Webhook processing failed. An error occurred."
}), 500
@connectors_bp.route("/api/connectors/webhooks/angi", methods=["POST"])
@limiter.limit("60 per minute")
def angi_webhook():
"""
Angi lead feed webhook endpoint — multi-tenant via x-api-key lookup.
Authenticates the caller by matching the `x-api-key` header to the
stored API key in a connector's encrypted config. This lets multiple
companies each have their own Angi connector on the same endpoint.
Expected payload:
{
"lead": { ... Angi lead fields ... }
}
Returns:
JSON response with processing status
"""
import logging
logger = logging.getLogger(__name__)
try:
# --- API key authentication ---
api_key = request.headers.get("x-api-key", "").strip()
if not api_key:
logger.warning("Angi webhook: Missing x-api-key header")
return jsonify({"success": False, "error": "Missing x-api-key"}), 401
# Find the connector whose config contains a matching api_key
connector = None
angi_connectors = Connector.query.filter_by(
service="angi",
status="connected",
).all()
for c in angi_connectors:
if c.config.get("api_key") == api_key:
connector = c
break
if not connector:
# Return 404 to avoid leaking whether a connector exists
logger.warning("Angi webhook: No connector matched x-api-key")
return jsonify({"success": False, "error": "Connector not found"}), 404
# --- Payload validation ---
payload = request.get_json(silent=True)
if not payload:
logger.warning("Angi webhook: Empty or invalid JSON payload")
return jsonify({"success": False, "error": "Invalid JSON payload"}), 400
# Build the connector instance
from app.connectors import build_connector
conn = build_connector(
service="angi",
company_id=connector.company_id,
config=connector.config,
connector_id=connector.id,
)
# Process the webhook
result = conn.process_webhook(
payload=payload,
request_headers=dict(request.headers),
)
if result.get("success"):
logger.info(
"Angi webhook: Lead %s processed for company %s",
result.get("lead_id", "unknown"),
connector.company_id,
)
return jsonify(result), 200
else:
logger.error(
"Angi webhook: Processing failed for company %s — %s",
connector.company_id,
result.get("error"),
)
# Return 200 to prevent Angi from retrying (data validation error)
return jsonify(result), 200
except Exception as e:
logger.error("Angi webhook processing error: %s", str(e), exc_info=True)
return jsonify({
"success": False,
"error": "Webhook processing failed.",
}), 200
# -- Zapier webhook endpoint ------------------------------------------------
@connectors_bp.route("/api/connectors/webhooks/zapier", methods=["POST"])
@limiter.limit("60 per minute")
def zapier_webhook():
"""
Zapier webhook endpoint.
Receives events from Zaps and dispatches them to the Zapier connector.
Expected payload:
{
"action": "create_lead" | "update_contact" | "log_activity",
"data": { ... fields for the specific action ... }
}
Returns:
JSON response with processing status
"""
import logging
logger = logging.getLogger(__name__)
try:
payload = request.get_json(silent=True)
if not payload:
logger.warning("Zapier webhook: Empty or invalid JSON payload")
return jsonify({
"success": False,
"error": "Invalid JSON payload"
}), 400
# Find the active Zapier connector
connector = Connector.query.filter_by(
service="zapier",
status="connected"
).first()
if not connector:
logger.warning("Zapier webhook: No active connector found")
# Return 200 to prevent Zapier from excessive retries
return jsonify({
"success": False,
"error": "No active Zapier connector configured"
}), 200
# Build the connector instance
from app.connectors import build_connector
conn = build_connector(
service="zapier",
company_id=connector.company_id,
config=connector.config,
connector_id=connector.id
)
# Process the webhook
result = conn.process_webhook(
payload=payload,
request_headers=dict(request.headers),
)
if result.get("success"):
action = result.get("action", "unknown")
logger.info(
"Zapier webhook: action='%s' processed successfully", action
)
return jsonify(result), 200
else:
logger.error(
"Zapier webhook: processing failed — %s",
result.get("error"),
)
# Return 200 to prevent Zapier from excessive retries on data errors
return jsonify(result), 200
except Exception as e:
logger.error(
"Zapier webhook processing error: %s", str(e), exc_info=True
)
return jsonify({
"success": False,
"error": "Webhook processing failed. An error occurred."
}), 200