"""Connector base framework — abstract base class and registry."""
from __future__ import annotations
import logging
import time
import threading
import secrets
import hashlib
import json
import redis as redis_lib
from abc import ABC, abstractmethod
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
import requests # noqa: F401 — needed by BaseConnector._is_transient
# -- Redis connection for cross-worker OAuth state ---------------------------
_redis = None
_redis_error = None
def _get_redis():
"""Lazily connect to Redis, caching the connection per-process."""
global _redis, _redis_error
if _redis is None and _redis_error is None:
try:
_redis = redis_lib.Redis(
host="127.0.0.1", port=6379, db=0,
socket_connect_timeout=1, decode_responses=True
)
_redis.ping()
except Exception as exc:
_redis_error = str(exc)
if _redis_error:
logger.debug("Redis unavailable for OAuth state: %s — using in-memory fallback", _redis_error)
return _redis
# Fallback in-memory dict when Redis is down
_OAUTH_STATE_FALLBACK: Dict[str, Dict[str, Any]] = {}
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Connector errors ----------------------------------------------------------
# ---------------------------------------------------------------------------
class ConnectorError(Exception):
"""Raised when a connector operation fails."""
pass
# ---------------------------------------------------------------------------
# Connector registry -------------------------------------------------------
# ---------------------------------------------------------------------------
_REGISTRY: Dict[str, type] = {}
_AVAILABLE_INFO: Dict[str, Dict[str, Any]] = {}
def register_connector(service: str, info: Dict[str, Any]) -> None:
"""Register a connector class with metadata for the /available endpoint."""
_AVAILABLE_INFO[service] = info
def get_available_connectors() -> List[Dict[str, Any]]:
"""Return a list of all registered connector metadata."""
return sorted(_AVAILABLE_INFO.values(), key=lambda c: c.get("service", ""))
def get_connector_class(service: str) -> Optional[type]:
"""Look up a connector class by service name."""
return _REGISTRY.get(service)
def get_connector_metadata(service: str) -> Optional[Dict[str, Any]]:
"""Look up connector metadata by service name."""
return _AVAILABLE_INFO.get(service)
def build_connector(
service: str,
company_id: str,
config: Dict[str, Any],
connector_id: Optional[str] = None,
) -> "BaseConnector":
"""Instantiate the right connector subclass for a service.
Raises:
ValueError: if the service is not registered.
"""
cls = get_connector_class(service)
if cls is None:
raise ValueError(f"Unknown connector service: {service}")
return cls(company_id=company_id, config=config, connector_id=connector_id)
# ---------------------------------------------------------------------------
# Base connector ------------------------------------------------------------
# ---------------------------------------------------------------------------
class BaseConnector(ABC):
"""Abstract base class for all integration connectors.
Subclasses implement ``connect()``, ``disconnect()``, ``sync()``, and
``status()``. The base class handles:
* Persistent logging to ``ConnectorLog`` on every operation.
* Retry logic with exponential back-off for transient failures.
* Rate-limit awareness (HTTP 429) with automatic back-off.
* A thread-safe ``_lock`` so concurrent calls on the same instance
don't corrupt state.
"""
# Class-level registry call (subclasses override _SERVICE and call this)
_SERVICE: str = ""
def __init__(
self,
*,
company_id: str,
config: Dict[str, Any],
connector_id: Optional[str] = None,
) -> None:
self.company_id = company_id
self.config = config or {}
self.connector_id = connector_id
self._connected = False
self._lock = threading.Lock()
# Retry defaults
self.max_retries: int = 3
self.base_backoff: float = 1.0 # seconds
# Rate-limit state (reset on every connect)
self._rate_limit_remaining = 0
self._rate_limit_reset_at: Optional[float] = None
# -- Lifecycle helpers --------------------------------------------------
def _log(
self,
event_type: str,
*,
status: str = "pending",
record_count: int = 0,
duration_ms: Optional[int] = None,
error_message: str = "",
details: Optional[Dict[str, Any]] = None,
) -> None:
"""Persist a log entry to the ``ConnectorLog`` model."""
try:
from app.models import db, ConnectorLog
# Skip log if we don't have a connector_id yet (pre-save phase)
if not self.connector_id:
logger.debug("Skipping log: no connector_id set yet")
return
log_entry = ConnectorLog(
company_id=self.company_id,
connector_id=self.connector_id,
event_type=event_type,
record_count=record_count,
duration_ms=duration_ms,
status=status,
error_message=error_message,
details_json=details or {},
)
db.session.add(log_entry)
db.session.commit()
except Exception:
db.session.rollback()
logger.exception("Failed to write connector log")
def _retry(self, func, *args, **kwargs):
"""Execute *func* with exponential back-off on transient errors.
Transient errors include:
* ConnectionError, TimeoutError
* HTTP 429 (rate limited) — honour Retry-After header when present.
* HTTP 5xx server errors.
Subclasses should mark their API wrapper methods to go through this.
"""
last_exc: Optional[Exception] = None
for attempt in range(1, self.max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as exc: # noqa: BLE001
last_exc = exc
if not self._is_transient(exc):
raise # Non-transient — give up immediately.
wait = self.base_backoff * (2 ** (attempt - 1))
logger.warning(
"Connector %s attempt %d/%d failed (%s), retrying in %.1fs",
self._SERVICE,
attempt,
self.max_retries,
exc,
wait,
)
time.sleep(wait)
raise last_exc # type: ignore[misc]
def _is_transient(self, exc: Exception) -> bool:
"""Heuristic: is this exception a candidate for retry?"""
# Never retry permanent blocks — "API access blocked" from Meta,
# 400 Bad Request, 401 Unauthorized, 403 Forbidden are NOT transient
if isinstance(exc, requests.HTTPError):
status = getattr(exc, "response", None)
if status:
code = getattr(status, "status_code", 0)
# 4xx errors are NOT transient (except 429)
if 400 <= code < 500 and code != 429:
return False
return code in (429, 500, 502, 503, 504)
if isinstance(exc, (ConnectionError, TimeoutError, OSError)):
return True
# Check for Meta's "API access blocked" in ValueError/ConnectorError
msg = str(exc)
if "API access blocked" in msg:
return False
return False
def _check_rate_limit(self) -> None:
"""Block briefly if we're inside a rate-limit window."""
if self._rate_limit_remaining <= 0 and self._rate_limit_reset_at:
wait = self._rate_limit_reset_at - time.time()
if wait > 0:
logger.info("Rate limit active, waiting %.1fs", wait)
time.sleep(wait)
self._rate_limit_remaining = 1 # Reset to allow 1 request
# -- Abstract methods ---------------------------------------------------
@abstractmethod
def connect(self) -> Dict[str, Any]:
"""Establish the connection (authenticate, validate credentials).
Returns:
dict with at least ``{'status': 'connected'}`` on success.
"""
...
@abstractmethod
def disconnect(self) -> Dict[str, Any]:
"""Tear down any active connection / revoke tokens."""
...
@abstractmethod
def sync(self) -> Dict[str, Any]:
"""Pull data from the external service and store locally.
Returns:
dict with ``record_count`` and ``status``.
"""
...
@abstractmethod
def status(self) -> Dict[str, Any]:
"""Check the current health of the connection."""
...
# ---------------------------------------------------------------------------
# OAuth 2.0 Connector ------------------------------------------------------
# ---------------------------------------------------------------------------
# Server-side OAuth state store — Redis-backed so Gunicorn workers
# share state. Keys: oauth_state:<hash> → json({state, expiry, ...})
# Falls back to in-memory dict if Redis is unreachable.
_OAUTH_STATE_TTL = 600 # seconds
class OAuthConnector(BaseConnector):
"""Base class for OAuth 2.0 connectors.
Adds ``authorize_url()``, ``exchange_code()``, and ``refresh_token()``
helpers on top of the standard connector lifecycle.
Subclasses must set class-level constants:
* ``OAUTH_AUTHORIZE_URL`` — provider's authorization endpoint
* ``OAUTH_TOKEN_URL`` — provider's token exchange endpoint
* ``OAUTH_SCOPES`` — list of scope strings to request
* ``OAUTH_CLIENT_ID`` — read from Flask config at runtime
and implement the two abstract methods below.
"""
OAUTH_AUTHORIZE_URL: str = ""
OAUTH_TOKEN_URL: str = ""
OAUTH_SCOPES: List[str] = []
OAUTH_CLIENT_ID: str = ""
OAUTH_CLIENT_SECRET: str = ""
REDIRECT_URI: str = ""
@abstractmethod
def oauth_authorize_url(self, state: str) -> str:
"""Build the provider's authorization URL with state parameter."""
...
@abstractmethod
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for access_token + refresh_token.
Returns a dict with at least ``access_token`` and ``refresh_token``,
plus any service-specific fields (realm_id, team_id, etc.).
Subclasses that use PKCE should accept ``pkce_verifier`` from kwargs.
"""
...
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh an expired access token using the standard OAuth 2.0 flow.
Override this method if your provider uses a non-standard refresh flow.
"""
if not self.OAUTH_TOKEN_URL or not refresh_token:
raise ValueError("Token URL or refresh_token not configured")
client_id = self._get_client_id()
client_secret = self._get_client_secret()
resp = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
timeout=30,
)
resp.raise_for_status()
return resp.json()
def _get_client_id(self) -> str:
"""Read the OAuth client ID from Flask config or class constant."""
from flask import current_app
cfg = current_app.config
prefix = f"OAUTH_{self._SERVICE.upper().replace('-', '_').replace('.', '_')}"
return cfg.get(f"{prefix}_CLIENT_ID", self.OAUTH_CLIENT_ID) or ""
def _get_client_secret(self) -> str:
"""Read the OAuth client secret from Flask config or class constant."""
from flask import current_app
cfg = current_app.config
prefix = f"OAUTH_{self._SERVICE.upper().replace('-', '_').replace('.', '_')}"
return cfg.get(f"{prefix}_CLIENT_SECRET", self.OAUTH_CLIENT_SECRET) or ""
def _get_redirect_uri(self) -> str:
"""Read the base redirect URI from Flask config."""
from flask import current_app
base = current_app.config.get("CONNECTOR_REDIRECT_URI", "")
return f"{base}/{self._SERVICE}" if base else ""
# ---------------------------------------------------------------------------
# OAuth state management (CSRF protection) ---------------------------------
# ---------------------------------------------------------------------------
def generate_oauth_state(service: str, company_id: str, connector_id: str = "", next_path: str = "", **extra) -> str:
"""Generate a CSRF-safe state token and store it server-side.
Returns the raw state string that must be passed to the provider's
authorize URL.
Extra keyword arguments (e.g. pkce_verifier) are stored in the state
dict for retrieval during the callback.
Stored in Redis for cross-worker sharing (Gunicorn). Falls back to
in-memory dict if Redis is unreachable.
"""
raw = secrets.token_urlsafe(32)
key = f"oauth_state:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"
state_data = {
"state": raw,
"service": service,
"company_id": company_id,
"connector_id": connector_id,
"next_path": next_path,
"expiry": (datetime.now(timezone.utc) + timedelta(seconds=_OAUTH_STATE_TTL)).isoformat(),
**extra,
}
r = _get_redis()
if r is not None:
r.setex(key, _OAUTH_STATE_TTL, json.dumps(state_data))
else:
# Fallback: in-memory
state_data["_expiry_dt"] = datetime.now(timezone.utc) + timedelta(seconds=_OAUTH_STATE_TTL)
_OAUTH_STATE_FALLBACK[key] = state_data
logger.info("OAuth state generated for %s (key=%s)", service, key.split(":")[1][:8])
return raw
def verify_oauth_state(service: str, state: str) -> Dict[str, Any]:
"""Verify and consume a CSRF state token.
Returns the stored metadata dict on success, raises ValueError on failure.
"""
now = datetime.now(timezone.utc)
# Try Redis first (scan for matching state)
r = _get_redis()
if r is not None:
# Scan all oauth_state: keys — there should be very few (<10)
cursor = 0
while True:
cursor, keys = r.scan(cursor, match="oauth_state:*", count=100)
for key in keys:
raw_json = r.get(key)
if raw_json:
data = json.loads(raw_json)
if data["state"] == state and data["service"] == service:
# Check expiry
expiry = datetime.fromisoformat(data["expiry"])
if now > expiry:
r.delete(key)
raise ValueError("OAuth state token expired")
r.delete(key) # consume
logger.info("OAuth state verified for %s (Redis)", service)
return data
# Fallback: in-memory dict
for key, data in list(_OAUTH_STATE_FALLBACK.items()):
if data["state"] == state and data["service"] == service:
expiry_dt = data.get("_expiry_dt", datetime.fromisoformat(data["expiry"]))
if now > expiry_dt:
del _OAUTH_STATE_FALLBACK[key]
raise ValueError("OAuth state token expired")
consumed = _OAUTH_STATE_FALLBACK.pop(key)
logger.info("OAuth state verified for %s (fallback)", service)
return consumed
raise ValueError("Invalid OAuth state token")
def _update_oauth_state(service: str, state: str, **kwargs) -> None:
"""Update fields on an existing OAuth state entry.
Used by Google Sheets to inject the PKCE verifier after it's generated.
"""
# Try Redis
r = _get_redis()
if r is not None:
cursor = 0
while True:
cursor, keys = r.scan(cursor, match="oauth_state:*", count=100)
for key in keys:
raw_json = r.get(key)
if raw_json:
data = json.loads(raw_json)
if data["state"] == state and data["service"] == service:
data.update(kwargs)
r.set(key, json.dumps(data))
logger.debug("OAuth state updated for %s: %s", service, list(kwargs.keys()))
return
if cursor == 0:
break
# Fallback: in-memory
for key, data in _OAUTH_STATE_FALLBACK.items():
if data["state"] == state and data["service"] == service:
data.update(kwargs)
logger.debug("OAuth state updated (fallback) for %s: %s", service, list(kwargs.keys()))
return
logger.warning("Could not update OAuth state for %s — not found", service)
def cleanup_expired_oauth_states() -> None:
"""Remove expired state tokens (call from cron or before each request)."""
now = datetime.now(timezone.utc)
# Clean Redis keys
r = _get_redis()
if r is not None:
cursor = 0
cleaned = 0
while True:
cursor, keys = r.scan(cursor, match="oauth_state:*", count=100)
for key in keys:
raw_json = r.get(key)
if raw_json:
data = json.loads(raw_json)
expiry = datetime.fromisoformat(data["expiry"])
if now > expiry:
r.delete(key)
cleaned += 1
if cursor == 0:
break
if cleaned:
logger.debug("Cleaned up %d expired OAuth states (Redis)", cleaned)
# Clean fallback in-memory dict
expired = [k for k, v in _OAUTH_STATE_FALLBACK.items()
if now > v.get("_expiry_dt", datetime.fromisoformat(v["expiry"]))]
for k in expired:
del _OAUTH_STATE_FALLBACK[k]
if expired:
logger.debug("Cleaned up %d expired OAuth states (fallback)", len(expired))
# ---------------------------------------------------------------------------
# Auto-import submodules so subclasses get registered at import time. -----
# This is deferred to avoid circular imports during app bootstrap.
# ---------------------------------------------------------------------------
def _import_submodules() -> None:
"""Import all connector implementations to trigger registration."""
# Import requests early so the _is_transient check works.
import requests # noqa: F401
from . import hubspot # noqa: F401
from . import quickbooks # noqa: F401
from . import google_ads # noqa: F401
from . import jobnimbus # noqa: F401
from . import slack # noqa: F401
from . import facebook_ads # noqa: F401
from . import generic_rest # noqa: F401
from . import servicetitan # noqa: F401
from . import angi # noqa: F401
from . import zapier # noqa: F401
from . import google_sheets # noqa: F401
from . import sms # noqa: F401
from . import five9 # noqa: F401
from . import leadperfection # noqa: F401
from . import marlimar # noqa: F401
from . import jobber # noqa: F401
logger.info("Registered connectors: %s", list(_AVAILABLE_INFO.keys()))