"""JobNimbus CRM connector.
Syncs contacts (people), deals, and companies from the JobNimbus REST API v4.
Paginates through endpoints using limit/skip, upserts records into the local
DB by external_id, and returns actual record counts.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import BaseConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata -------------------------------------------------
register_connector(
"jobnimbus",
{
"service": "jobnimbus",
"name": "JobNimbus",
"category": "crm",
"description": "CRM for home service companies — contacts, deals, companies.",
"auth_type": "api_key",
"auth_fields": ["api_key"],
"capabilities": ["contacts", "deals", "companies"],
"rate_limit": "Not documented; conservative back-off applied.",
"docs_url": "https://developer.jobnimbus.com/",
},
)
# -- Internal API helpers ---------------------------------------------------
def _api_get(url: str, api_key: str) -> Dict[str, Any]:
"""Make a GET request to JobNimbus API with X-API-Key header."""
headers = {"X-API-Key": api_key}
resp = requests.get(url, headers=headers, timeout=30)
# Capture rate-limit headers
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
logger.warning(
"JobNimbus rate limited: status=429, Retry-After=%s",
retry_after,
)
resp.raise_for_status()
return resp.json()
def _safe_float(value: Any) -> Optional[float]:
"""Convert a value to float, returning None on failure."""
if not value:
return None
try:
return float(value)
except (ValueError, TypeError):
return None
# -- Connector class --------------------------------------------------------
class JobNimbusConnector(BaseConnector):
"""JobNimbus CRM integration — pulls people (contacts), deals, companies."""
_SERVICE = "jobnimbus"
_BASE_URL = "https://api.jobnimbus.com/api/v4"
PAGE_LIMIT = 100
# -- connect / disconnect -----------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate the API key by fetching account info."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
api_key = self.config.get("api_key", "")
if not api_key:
raise ValueError("JobNimbus api_key is required")
account_info = self._retry(
_api_get, f"{self._BASE_URL}/account/info", api_key
)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"account_info": account_info},
)
return {
"status": "connected",
"service": "jobnimbus",
"account_info": account_info,
"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]:
"""Clear the connection flag and log the event."""
self._connected = False
self.config.pop("api_key", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "jobnimbus"}
# -- sync ----------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync contacts (people), companies, and deals from JobNimbus."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
# Sync contacts first (deals reference them)
contacts = self._sync_contacts()
total_records += contacts
results["contacts"] = contacts
# Sync companies
companies = self._sync_companies()
total_records += companies
results["companies"] = companies
# Sync deals (they reference contacts)
deals = self._sync_deals()
total_records += deals
results["deals"] = deals
duration_ms = int((time.monotonic() - start) * 1000)
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:
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,
}
# -- Paginated fetch helper ---------------------------------------------
def _fetch_all(self, endpoint: str) -> List[Dict[str, Any]]:
"""Paginate through a JobNimbus endpoint using limit/skip.
JobNimbus returns a dict with 'data' (list) and 'pagination' info.
"""
api_key = self.config.get("api_key", "")
if not api_key:
raise ValueError("No api_key configured for JobNimbus")
all_items: List[Dict[str, Any]] = []
offset = 0
while True:
url = f"{self._BASE_URL}/{endpoint}"
params = {"limit": str(self.PAGE_LIMIT), "skip": str(offset)}
self._check_rate_limit()
resp = self._retry(_api_get, url, api_key)
# JobNimbus returns a dict; the list lives under 'data'
data = resp.get("data", []) if isinstance(resp, dict) else []
if not data:
break
all_items.extend(data)
logger.info(
"JobNimbus %s: fetched %d (total %d)",
endpoint, len(data), len(all_items),
)
# Check if we have more pages
total = resp.get("pagination", {}).get("total", 0) if isinstance(resp, dict) else 0
if len(all_items) >= total or len(data) < self.PAGE_LIMIT:
break
offset += len(data)
time.sleep(0.1) # Polite pause between pages
logger.info("JobNimbus %s: total %d records", endpoint, len(all_items))
return all_items
# -- Merge (upsert) helper -----------------------------------------------
@staticmethod
def _merge(model, company_id: str, external_id: str, defaults: Dict[str, Any]) -> int:
"""Find or create a record by (company_id, external_id) and update fields.
Returns 1 if a record was inserted or updated.
"""
from app.models import db
record = model.query.filter_by(
company_id=company_id, external_id=external_id
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults,
)
db.session.add(record)
return 1
# -- Contact (People) sync -----------------------------------------------
def _sync_contacts(self) -> int:
"""Pull people from JobNimbus, merge into local CrmContact table."""
from app.models import db, CrmContact
items = self._fetch_all("people")
count = 0
for person in items:
person_id = person.get("ID", "") or person.get("PersonID", "")
if not person_id:
continue
# JobNimbus address fields
address = person.get("AddressLine1", "")
city = person.get("City", "")
state = person.get("State", "")
zipcode = person.get("Zip", "")
count += self._merge(
CrmContact,
self.company_id,
str(person_id),
{
"email": person.get("Email", ""),
"first_name": person.get("FirstName", ""),
"last_name": person.get("LastName", ""),
"phone": person.get("Phone", ""),
"company_name": person.get("CompanyName", ""),
"lifecycle_stage": person.get("Status", ""),
"hubspot_owner_id": person.get("OwnerID", ""),
"properties_json": {
"address": address,
"city": city,
"state": state,
"zip": zipcode,
"source": person.get("Source", ""),
"status": person.get("Status", ""),
"owner_id": person.get("OwnerID", ""),
},
},
)
db.session.commit()
logger.info("JobNimbus contacts synced: %d records merged", count)
return count
# -- Deal sync -----------------------------------------------------------
def _sync_deals(self) -> int:
"""Pull deals from JobNimbus, merge into local CrmDeal table."""
from app.models import db, CrmDeal, CrmContact
items = self._fetch_all("deals")
count = 0
for deal in items:
deal_id = deal.get("ID", "")
if not deal_id:
continue
# Parse amount
amount_raw = deal.get("Amount")
amount = _safe_float(amount_raw)
# Parse expected close date
close_date_str = deal.get("CloseDate", "") or deal.get("TargetDate", "")
expected_close_date: Optional[datetime] = None
if close_date_str:
for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
try:
expected_close_date = datetime.strptime(
close_date_str[:19], fmt[:19] if "T" in fmt else fmt
)
break
except (ValueError, TypeError):
continue
# Parse probability from stage
stage = deal.get("Status", "")
probability: Optional[float] = None
if stage:
stage_lower = stage.lower()
if "won" in stage_lower:
probability = 1.0
elif "lost" in stage_lower or "cancelled" in stage_lower:
probability = 0.0
# Try to link to contact via PersonID
person_id = deal.get("PersonID", "") or deal.get("PersonId", "")
contact_id: Optional[str] = None
if person_id:
contact = CrmContact.query.filter_by(
company_id=self.company_id,
external_id=str(person_id),
).first()
if contact:
contact_id = contact.id
count += self._merge(
CrmDeal,
self.company_id,
str(deal_id),
{
"name": deal.get("Name", ""),
"amount": amount,
"stage": stage,
"pipeline": "",
"probability": probability,
"expected_close_date": expected_close_date,
"contact_id": contact_id,
"deal_owner_id": deal.get("OwnerID", ""),
"properties_json": {
"source": deal.get("Source", ""),
"status": deal.get("Status", ""),
"owner_id": deal.get("OwnerID", ""),
"person_id": person_id,
},
},
)
db.session.commit()
logger.info("JobNimbus deals synced: %d records merged", count)
return count
# -- Company sync --------------------------------------------------------
def _sync_companies(self) -> int:
"""Pull companies from JobNimbus, merge into local CrmCompany table."""
from app.models import db, CrmCompany
items = self._fetch_all("companies")
count = 0
for company in items:
company_id = company.get("ID", "")
if not company_id:
continue
count += self._merge(
CrmCompany,
self.company_id,
str(company_id),
{
"name": company.get("Name", ""),
"domain": company.get("Website", "").replace("https://", "").replace("http://", "").rstrip("/"),
"industry": company.get("Category", ""),
"num_employees": None,
"properties_json": {
"phone": company.get("Phone", ""),
"address": company.get("AddressLine1", ""),
"city": company.get("City", ""),
"state": company.get("State", ""),
"zip": company.get("Zip", ""),
"website": company.get("Website", ""),
"category": company.get("Category", ""),
},
},
)
db.session.commit()
logger.info("JobNimbus companies synced: %d records merged", count)
return count
# -- status --------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Check current health of the JobNimbus connection."""
if not self._connected:
return {
"service": "jobnimbus",
"connected": False,
"config_present": bool(self.config.get("api_key")),
}
try:
api_key = self.config.get("api_key", "")
if not api_key:
return {
"service": "jobnimbus",
"connected": False,
"error": "No api_key configured",
}
account_info = self._retry(
_api_get, f"{self._BASE_URL}/account/info", api_key
)
return {
"service": "jobnimbus",
"connected": True,
"account_info": account_info,
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {"service": "jobnimbus", "connected": False, "error": str(exc)}
# -- Register in the framework registry -------------------------------------
_REGISTRY["jobnimbus"] = JobNimbusConnector