"""Generic REST API connector — connect to any REST endpoint with configurable auth and pagination."""
from __future__ import annotations
import base64
import logging
import time
from typing import Any, Dict, List, Optional
import requests
from . import _REGISTRY, register_connector
logger = logging.getLogger(__name__)
class GenericRestConnector:
"""Connector for arbitrary REST APIs.
Config keys (all values come from *config* dict):
Auth
auth_type — ``'none' | 'api_key' | 'basic' | 'bearer'``
api_key — value for API-key or bearer auth
api_key_header — header name for API-key auth (default ``X-API-Key``)
basic_user — username for basic auth
basic_password — password for basic auth
Endpoint
endpoint — base URL for data fetches (required)
health_endpoint — URL for status / health checks
Pagination
pagination_type — ``'offset' | 'page' | 'cursor' | 'link' | 'none'``
page_size — items per page / offset step (default 100)
offset_param — query param name for offset (default ``offset``)
limit_param — query param name for limit (default ``limit``)
page_param — query param name for page (default ``page``)
per_page_param — query param name for per-page (default ``per_page``)
cursor_param — query param name for cursor (default ``cursor``)
next_link_key — JSON key / path holding the next-link URL (default ``next_url``)
Data mapping
data_key — JSON key holding the list of records (default ``results``)
id_key — JSON key for the unique identifier (default ``id``)
entity_type — local entity label used in upsert / logging
Rate-limiting
rate_limit_remaining_header — response header for remaining quota
rate_limit_reset_header — response header for reset timestamp
"""
_SERVICE = "generic_rest"
# ------------------------------------------------------------------
# Init & registration
# ------------------------------------------------------------------
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._session: Optional[requests.Session] = None
# Retry defaults (same as BaseConnector)
self.max_retries: int = 3
self.base_backoff: float = 1.0
# Rate-limit state
self._rate_limit_remaining = 0
self._rate_limit_reset_at: Optional[float] = None
# Connection metadata (populated by connect / sync)
self._account_info: Dict[str, Any] = {}
self._last_sync: Optional[str] = None
# ------------------------------------------------------------------
# Public API (implements BaseConnector interface)
# ------------------------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate credentials against the configured endpoint.
Sends a lightweight GET to *endpoint* with configured auth to
confirm the credentials work. Does **not** pull data yet.
Returns ``{'status': 'connected', 'account_info': ...}`` on success.
"""
endpoint = self.config.get("endpoint")
if not endpoint:
raise ValueError("Config requires 'endpoint'")
self._connected = False
self._session = requests.Session()
try:
resp = self._retry(
self._session.get,
endpoint,
headers=self._auth_headers(),
timeout=15,
)
resp.raise_for_status()
except requests.HTTPError as exc:
self._connected = False
self._session.close()
self._session = None
raise ConnectionError(
f"Generic REST connect failed: {exc.response.status_code} {exc.response.text}"
) from exc
self._connected = True
self._capture_rate_limit(resp)
# Try to extract account info from response JSON (best-effort)
self._account_info = self._extract_account_info(resp)
self._log(
"connect",
status="success",
details={"endpoint": endpoint},
)
return {
"status": "connected",
"account_info": self._account_info,
}
# ------------------------------------------------------------------
def disconnect(self) -> Dict[str, Any]:
"""Close session and clear in-memory credentials."""
if self._session:
self._session.close()
self._session = None
self._connected = False
self._rate_limit_remaining = 0
self._rate_limit_reset_at = None
self._account_info = {}
# Wipe secrets from the config dict so they aren't leaked
for key in ("api_key", "basic_password", "bearer_token"):
self.config.pop(key, None)
self._log("disconnect", status="success")
return {"status": "disconnected"}
# ------------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Paginate through the configured endpoint and upsert records.
Supports offset, page, cursor, and link-based pagination.
Returns ``{'record_count': N, 'status': 'success'}``.
"""
if not self._connected:
raise RuntimeError("Not connected — call connect() first")
if not self._session:
raise RuntimeError("Session not initialised")
endpoint = self.config.get("endpoint")
if not endpoint:
raise ValueError("Config requires 'endpoint'")
start_time = time.time()
pagination_type = self.config.get("pagination_type", "none")
entity_type = self.config.get("entity_type", "record")
data_key = self.config.get("data_key", "results")
id_key = self.config.get("id_key", "id")
all_records: List[Dict[str, Any]] = []
if pagination_type == "offset":
all_records = self._paginate_offset(
endpoint, data_key, id_key
)
elif pagination_type == "page":
all_records = self._paginate_page(
endpoint, data_key, id_key
)
elif pagination_type == "cursor":
all_records = self._paginate_cursor(
endpoint, data_key, id_key
)
elif pagination_type == "link":
all_records = self._paginate_link(
endpoint, data_key, id_key
)
else:
# Single-page fetch
all_records = self._fetch_page(endpoint, data_key, id_key)
duration_ms = int((time.time() - start_time) * 1000)
# Upsert into local DB
upserted = self._upsert_records(all_records, entity_type, id_key)
self._last_sync = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
self._log(
"sync",
status="success",
record_count=len(upserted),
duration_ms=duration_ms,
details={
"pagination_type": pagination_type,
"entity_type": entity_type,
"fetched": len(all_records),
"upserted": len(upserted),
},
)
return {
"status": "success",
"record_count": len(upserted),
"fetched": len(all_records),
"duration_ms": duration_ms,
}
# ------------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Check connection health via the configured health endpoint."""
if not self._connected or not self._session:
return {
"status": "disconnected",
"connected": False,
"last_sync": self._last_sync,
}
health_endpoint = self.config.get("health_endpoint")
if not health_endpoint:
# Fall back to the data endpoint with a HEAD
health_endpoint = self.config.get("endpoint")
if health_endpoint:
try:
resp = self._retry(
self._session.get,
health_endpoint,
headers=self._auth_headers(),
timeout=10,
)
healthy = resp.status_code < 400
self._capture_rate_limit(resp)
except Exception as exc:
return {
"status": "error",
"connected": self._connected,
"last_sync": self._last_sync,
"error": str(exc),
}
else:
healthy = True
return {
"status": "healthy" if healthy else "degraded",
"connected": self._connected,
"last_sync": self._last_sync,
"account_info": self._account_info,
}
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
# -- Logging / retry (mirrors BaseConnector interface) ----------------
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
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."""
last_exc: Optional[Exception] = None
for attempt in range(1, self.max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as exc:
last_exc = exc
if not self._is_transient(exc):
raise
wait = self.base_backoff * (2 ** (attempt - 1))
logger.warning(
"Generic REST attempt %d/%d failed (%s), retrying in %.1fs",
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?"""
if isinstance(exc, (ConnectionError, TimeoutError, OSError)):
return True
if isinstance(exc, requests.HTTPError):
resp = getattr(exc, "response", None)
if resp:
code = getattr(resp, "status_code", 0)
return code in (429, 500, 502, 503, 504)
return False
def _capture_rate_limit(self, resp: requests.Response) -> None:
"""Read rate-limit headers and update back-off state."""
remaining_header = self.config.get(
"rate_limit_remaining_header", "X-RateLimit-Remaining"
)
reset_header = self.config.get(
"rate_limit_reset_header", "X-RateLimit-Reset"
)
remaining = resp.headers.get(remaining_header)
if remaining:
try:
self._rate_limit_remaining = int(remaining)
except (ValueError, TypeError):
pass
reset_val = resp.headers.get(reset_header)
if reset_val:
try:
self._rate_limit_reset_at = float(reset_val)
except (ValueError, TypeError):
pass
# Also handle standard 429 Retry-After
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
if retry_after:
try:
self._rate_limit_reset_at = time.time() + float(retry_after)
self._rate_limit_remaining = 0
except (ValueError, TypeError):
pass
# -- Auth -------------------------------------------------------------
def _auth_headers(self) -> Dict[str, str]:
"""Build auth headers based on config."""
auth_type = self.config.get("auth_type", "none")
if auth_type == "api_key":
header_name = self.config.get("api_key_header", "X-API-Key")
api_key = self.config.get("api_key", "")
return {header_name: api_key}
elif auth_type == "basic":
user = self.config.get("basic_user", "")
pwd = self.config.get("basic_password", "")
encoded = base64.b64encode(f"{user}:{pwd}".encode()).decode()
return {"Authorization": f"Basic {encoded}"}
elif auth_type == "bearer":
token = self.config.get("bearer_token", self.config.get("api_key", ""))
return {"Authorization": f"Bearer {token}"}
# auth_type == "none" or anything unrecognised
return {}
# -- Pagination -------------------------------------------------------
def _fetch_page(
self,
url: str,
data_key: str,
id_key: str,
params: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
"""Fetch a single page and return extracted records."""
resp = self._retry(
self._session.get,
url,
headers=self._auth_headers(),
params=params or {},
timeout=30,
)
resp.raise_for_status()
self._capture_rate_limit(resp)
body = resp.json()
records = body.get(data_key, body if isinstance(body, list) else [])
if not isinstance(records, list):
records = [records]
return self._normalize_records(records, id_key)
def _paginate_offset(
self,
endpoint: str,
data_key: str,
id_key: str,
) -> List[Dict[str, Any]]:
"""Offset/limit pagination."""
all_records: List[Dict[str, Any]] = []
offset_param = self.config.get("offset_param", "offset")
limit_param = self.config.get("limit_param", "limit")
page_size = self.config.get("page_size", 100)
offset = 0
while True:
params = {
offset_param: offset,
limit_param: page_size,
}
records = self._fetch_page(endpoint, data_key, id_key, params)
if not records:
break
all_records.extend(records)
offset += page_size
if len(records) < page_size:
break
return all_records
def _paginate_page(
self,
endpoint: str,
data_key: str,
id_key: str,
) -> List[Dict[str, Any]]:
"""Page/per_page pagination."""
all_records: List[Dict[str, Any]] = []
page_param = self.config.get("page_param", "page")
per_page_param = self.config.get("per_page_param", "per_page")
page_size = self.config.get("page_size", 100)
page = 1
while True:
params = {
page_param: page,
per_page_param: page_size,
}
records = self._fetch_page(endpoint, data_key, id_key, params)
if not records:
break
all_records.extend(records)
page += 1
if len(records) < page_size:
break
return all_records
def _paginate_cursor(
self,
endpoint: str,
data_key: str,
id_key: str,
) -> List[Dict[str, Any]]:
"""Cursor-based pagination."""
all_records: List[Dict[str, Any]] = []
cursor_param = self.config.get("cursor_param", "cursor")
limit_param = self.config.get("limit_param", "limit")
page_size = self.config.get("page_size", 100)
cursor: Optional[str] = None
while True:
params: Dict[str, Any] = {}
if cursor is not None:
params[cursor_param] = cursor
else:
params[limit_param] = page_size
resp = self._retry(
self._session.get,
endpoint,
headers=self._auth_headers(),
params=params,
timeout=30,
)
resp.raise_for_status()
self._capture_rate_limit(resp)
body = resp.json()
records = body.get(data_key, body if isinstance(body, list) else [])
if not isinstance(records, list):
records = [records]
if not records:
break
all_records.extend(self._normalize_records(records, id_key))
# Try common cursor next-token keys
cursor = (
body.get("next_cursor")
or body.get("nextCursor")
or body.get("next_page")
)
if not body.get("has_more", bool(cursor)):
break
return all_records
def _paginate_link(
self,
endpoint: str,
data_key: str,
id_key: str,
) -> List[Dict[str, Any]]:
"""Link-based pagination (follows next_url from response)."""
all_records: List[Dict[str, Any]] = []
next_link_key = self.config.get("next_link_key", "next_url")
url = endpoint
while url:
resp = self._retry(
self._session.get,
url,
headers=self._auth_headers(),
timeout=30,
)
resp.raise_for_status()
self._capture_rate_limit(resp)
body = resp.json()
records = body.get(data_key, body if isinstance(body, list) else [])
if not isinstance(records, list):
records = [records]
if not records:
break
all_records.extend(self._normalize_records(records, id_key))
# Determine next URL: from body key, then Link header
url = body.get(next_link_key) or body.get("next")
if not url:
link_header = resp.headers.get("Link", "")
if 'rel="next"' in link_header:
for part in link_header.split(","):
if 'rel="next"' in part or "rel='next'" in part:
url = part.split(";")[0].strip().strip("<>")
break
else:
url = None
return all_records
# -- Record normalisation & upsert ------------------------------------
@staticmethod
def _normalize_records(
records: List[Dict[str, Any]],
id_key: str,
) -> List[Dict[str, Any]]:
"""Ensure every record has an 'id' field (from id_key)."""
normalised: List[Dict[str, Any]] = []
for rec in records:
if not isinstance(rec, dict):
continue
if id_key in rec and "id" not in rec:
rec["id"] = rec.pop(id_key)
normalised.append(rec)
return normalised
def _upsert_records(
self,
records: List[Dict[str, Any]],
entity_type: str,
id_key: str,
) -> List[Dict[str, Any]]:
"""Upsert records into the local integration_data table.
Uses a generic key-value approach: each record is stored as JSON
keyed by entity_type + record id.
"""
if not records:
return []
try:
from app.models import db, IntegrationData
upserted: List[Dict[str, Any]] = []
for rec in records:
record_id = str(rec.get(id_key, rec.get("id", "")))
if not record_id:
logger.warning("Skipping record without id: %s", rec)
continue
data_entry = IntegrationData(
company_id=self.company_id,
source=self._SERVICE,
entity_type=entity_type,
entity_id=record_id,
data_json=rec,
)
db.session.merge(data_entry)
upserted.append(rec)
db.session.commit()
return upserted
except Exception:
db.session.rollback()
logger.exception("Failed to upsert records")
raise
# -- Account info extraction -----------------------------------------
@staticmethod
def _extract_account_info(resp: requests.Response) -> Dict[str, Any]:
"""Best-effort extraction of account info from response."""
try:
body = resp.json()
if isinstance(body, dict):
# Try common account info keys
for key in ("account", "user", "profile", "data"):
if key in body:
return body[key]
# Fall back to first-level dict (sanitised)
safe = dict(body)
for k in ("results", "data", "records", "items"):
safe.pop(k, None)
if safe:
return safe
except Exception:
pass
return {}
# ---------------------------------------------------------------------------
# Registration (mirrors pattern in hubspot.py / quickbooks.py)
# ---------------------------------------------------------------------------
def _register() -> None:
"""Register this connector with the framework."""
_REGISTRY["generic_rest"] = GenericRestConnector
register_connector(
"generic_rest",
{
"service": "generic_rest",
"name": "Generic REST API",
"description": (
"Connect to any REST API endpoint with configurable "
"authentication (API key, Basic, Bearer) and pagination "
"(offset, cursor, page, link)."
),
"auth_type": "configurable",
"config_schema": {
"endpoint": {
"type": "string",
"required": True,
"description": "Base URL for data fetches",
},
"auth_type": {
"type": "string",
"required": True,
"choices": ["none", "api_key", "basic", "bearer"],
"description": "Authentication method",
},
"api_key": {
"type": "string",
"sensitive": True,
"description": "API key or bearer token value",
},
"api_key_header": {
"type": "string",
"default": "X-API-Key",
"description": "Header name for API-key auth",
},
"basic_user": {
"type": "string",
"description": "Username for basic auth",
},
"basic_password": {
"type": "string",
"sensitive": True,
"description": "Password for basic auth",
},
"bearer_token": {
"type": "string",
"sensitive": True,
"description": "Bearer token for OAuth2",
},
"pagination_type": {
"type": "string",
"choices": ["none", "offset", "page", "cursor", "link"],
"default": "offset",
"description": "Pagination strategy",
},
"page_size": {
"type": "integer",
"default": 100,
"description": "Items per page",
},
"data_key": {
"type": "string",
"default": "results",
"description": "JSON key holding the list of records",
},
"id_key": {
"type": "string",
"default": "id",
"description": "JSON key for unique identifier",
},
"entity_type": {
"type": "string",
"default": "record",
"description": "Entity type label for local storage",
},
"health_endpoint": {
"type": "string",
"description": "URL for health/status checks",
},
},
"capabilities": ["read"],
},
)
logger.info("Registered generic_rest connector")
# Auto-register on import
_register()