8eb814283d
- Fix create_time_entry() using self._client instead of self.client - GET /member-mappings now returns all active account users, not just mapped ones — allows manual assignment when auto-match by email doesn't work - PsaMemberMappingResponse mapping fields are now Optional (id, external_member_id, external_member_name, matched_by) to represent unmapped users - Frontend MemberMappingTab skips null external_member_id when building localMappings, and derives user list from all returned entries - Add docs/connectwise-psa-testing-checklist.md Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
567 lines
22 KiB
Python
567 lines
22 KiB
Python
"""ConnectWise implementation of PSAProvider."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from app.services.psa.base import PSAProvider
|
|
from app.services.psa.cache import psa_cache
|
|
from app.services.psa.types import (
|
|
ConnectionTestResult,
|
|
PSATicket,
|
|
PSANote,
|
|
PSAStatus,
|
|
PSACompany,
|
|
PSAMember,
|
|
PSAConfiguration,
|
|
PSATimeEntry,
|
|
)
|
|
from .client import ConnectWiseClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConnectWiseProvider(PSAProvider):
|
|
"""ConnectWise PSA provider implementation."""
|
|
|
|
def __init__(self, client: ConnectWiseClient):
|
|
self.client = client
|
|
|
|
async def test_connection(self) -> ConnectionTestResult:
|
|
"""Test the CW connection by fetching system info."""
|
|
try:
|
|
info = await self.client.get("/system/info")
|
|
return ConnectionTestResult(
|
|
success=True,
|
|
message="Connected successfully.",
|
|
server_version=info.get("version", None),
|
|
)
|
|
except Exception as e:
|
|
return ConnectionTestResult(
|
|
success=False,
|
|
message=str(e),
|
|
server_version=None,
|
|
)
|
|
|
|
# ── Tickets ───────────────────────────────────────────────────────
|
|
|
|
async def get_ticket(self, ticket_id: str) -> PSATicket:
|
|
"""Fetch a single ticket by ID from ConnectWise."""
|
|
data = await self.client.get(
|
|
f"/service/tickets/{ticket_id}",
|
|
params={"fields": "id,summary,company,board,status,priority,closedFlag"},
|
|
)
|
|
return self._map_ticket(data)
|
|
|
|
async def search_tickets(self, query: str, **filters) -> list[PSATicket]:
|
|
"""Search CW tickets by summary. Supports board_id and status_id filters."""
|
|
params: dict = {
|
|
"fields": "id,summary,company,board,status,priority,closedFlag",
|
|
"orderBy": "id desc",
|
|
"pageSize": 25,
|
|
}
|
|
|
|
# Build CW condition query
|
|
conditions: list[str] = []
|
|
if query:
|
|
conditions.append(f"summary contains '{query}'")
|
|
if filters.get("board_id"):
|
|
conditions.append(f"board/id = {filters['board_id']}")
|
|
if filters.get("status_id"):
|
|
conditions.append(f"status/id = {filters['status_id']}")
|
|
if not filters.get("include_closed", False):
|
|
conditions.append("closedFlag = false")
|
|
|
|
if conditions:
|
|
params["conditions"] = " and ".join(conditions)
|
|
|
|
data = await self.client.get("/service/tickets", params=params)
|
|
|
|
return [
|
|
self._map_ticket(t)
|
|
for t in (data if isinstance(data, list) else [])
|
|
]
|
|
|
|
async def get_ticket_configurations(
|
|
self, ticket_id: str
|
|
) -> list[PSAConfiguration]:
|
|
"""Get configurations (assets) attached to a ticket."""
|
|
data = await self.client.get(
|
|
f"/service/tickets/{ticket_id}/configurations",
|
|
params={"fields": "id,deviceIdentifier,type,company"},
|
|
)
|
|
return [
|
|
PSAConfiguration(
|
|
id=str(c["id"]),
|
|
name=c.get("deviceIdentifier", ""),
|
|
type=c.get("type", {}).get("name") if c.get("type") else None,
|
|
company_name=c.get("company", {}).get("name") if c.get("company") else None,
|
|
)
|
|
for c in (data if isinstance(data, list) else [])
|
|
]
|
|
|
|
# ── Board statuses (cached) ───────────────────────────────────────
|
|
|
|
async def get_ticket_statuses(self, board_id: int) -> list[PSAStatus]:
|
|
"""Get available statuses for a CW service board (cached 1 hour)."""
|
|
cache_key = f"board_statuses:{board_id}"
|
|
cached = psa_cache.get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
data = await self.client.get(
|
|
f"/service/boards/{board_id}/statuses",
|
|
params={"fields": "id,name,closedStatus", "pageSize": 100},
|
|
)
|
|
result = [
|
|
PSAStatus(
|
|
id=s["id"],
|
|
name=s["name"],
|
|
is_closed=s.get("closedStatus", False),
|
|
)
|
|
for s in (data if isinstance(data, list) else [])
|
|
]
|
|
psa_cache.set(cache_key, result, ttl_seconds=3600)
|
|
return result
|
|
|
|
# ── Companies ─────────────────────────────────────────────────────
|
|
|
|
async def list_companies(self, **filters) -> list[PSACompany]:
|
|
"""List companies from CW, optionally filtered by status."""
|
|
params: dict = {
|
|
"fields": "id,name,status",
|
|
"pageSize": 100,
|
|
"orderBy": "name asc",
|
|
}
|
|
conditions: list[str] = []
|
|
if filters.get("status"):
|
|
conditions.append(f"status/name = '{filters['status']}'")
|
|
if conditions:
|
|
params["conditions"] = " and ".join(conditions)
|
|
|
|
data = await self.client.get("/company/companies", params=params)
|
|
return [
|
|
PSACompany(
|
|
id=str(c["id"]),
|
|
name=c.get("name", ""),
|
|
status=c.get("status", {}).get("name") if c.get("status") else None,
|
|
)
|
|
for c in (data if isinstance(data, list) else [])
|
|
]
|
|
|
|
async def get_company(self, company_id: str) -> PSACompany:
|
|
"""Fetch a single company by ID."""
|
|
data = await self.client.get(
|
|
f"/company/companies/{company_id}",
|
|
params={"fields": "id,name,status"},
|
|
)
|
|
return PSACompany(
|
|
id=str(data["id"]),
|
|
name=data.get("name", ""),
|
|
status=data.get("status", {}).get("name") if data.get("status") else None,
|
|
)
|
|
|
|
# ── Notes & status updates ───────────────────────────────────────
|
|
|
|
async def post_note(
|
|
self,
|
|
ticket_id: str,
|
|
text: str,
|
|
note_type: str,
|
|
member_id: str | None = None,
|
|
) -> PSANote:
|
|
"""Post a note to a CW ticket.
|
|
|
|
Maps ResolutionFlow note types to CW flag fields:
|
|
- internal_analysis → internalAnalysisFlag (internal only)
|
|
- resolution → resolutionFlag (internal, triggers notifications)
|
|
- description → detailDescriptionFlag (external, triggers notifications)
|
|
"""
|
|
from app.services.psa.types import NoteType
|
|
|
|
flags = {
|
|
NoteType.INTERNAL_ANALYSIS: {
|
|
"internalAnalysisFlag": True,
|
|
"resolutionFlag": False,
|
|
"detailDescriptionFlag": False,
|
|
"internalFlag": True,
|
|
"processNotifications": False,
|
|
},
|
|
NoteType.RESOLUTION: {
|
|
"internalAnalysisFlag": False,
|
|
"resolutionFlag": True,
|
|
"detailDescriptionFlag": False,
|
|
"internalFlag": True,
|
|
"processNotifications": True,
|
|
},
|
|
NoteType.DESCRIPTION: {
|
|
"internalAnalysisFlag": False,
|
|
"resolutionFlag": False,
|
|
"detailDescriptionFlag": True,
|
|
"internalFlag": False,
|
|
"processNotifications": True,
|
|
},
|
|
}
|
|
|
|
note_flags = flags.get(note_type, flags[NoteType.INTERNAL_ANALYSIS])
|
|
|
|
# NOTE: CW Developer Guide states \n is "Not Supported" in JSON bodies
|
|
# and may be collapsed to a single space. CW does support markdown in ticket
|
|
# notes (see PSA-Markdown.md). This needs sandbox testing — if newlines are
|
|
# lost, consider using double-space line breaks or HTML <br> tags instead.
|
|
body: dict = {
|
|
"text": text,
|
|
**note_flags,
|
|
}
|
|
|
|
if member_id:
|
|
body["member"] = {"id": int(member_id)}
|
|
|
|
data = await self.client.post(
|
|
f"/service/tickets/{ticket_id}/notes", json_body=body
|
|
)
|
|
|
|
return PSANote(
|
|
id=str(data.get("id", "")),
|
|
text=data.get("text", ""),
|
|
note_type=note_type,
|
|
created_at=data.get("dateCreated"),
|
|
)
|
|
|
|
async def update_ticket_status(
|
|
self, ticket_id: str, status_id: int
|
|
) -> PSATicket:
|
|
"""Update a CW ticket's status using JSON Patch format."""
|
|
patch_body = [
|
|
{"op": "replace", "path": "status", "value": {"id": status_id}}
|
|
]
|
|
data = await self.client.patch(
|
|
f"/service/tickets/{ticket_id}", json_body=patch_body
|
|
)
|
|
return self._map_ticket(data)
|
|
|
|
async def list_members(self) -> list[PSAMember]:
|
|
"""List CW system members (cached 15 minutes)."""
|
|
cache_key = "members:all"
|
|
cached = psa_cache.get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
data = await self.client.get_paginated(
|
|
"/system/members",
|
|
params={
|
|
"fields": "id,identifier,firstName,lastName,officeEmail",
|
|
"conditions": "inactiveFlag = false",
|
|
"pageSize": 1000,
|
|
},
|
|
)
|
|
|
|
result = [
|
|
PSAMember(
|
|
id=str(m["id"]),
|
|
identifier=m.get("identifier", ""),
|
|
name=f"{m.get('firstName', '')} {m.get('lastName', '')}".strip(),
|
|
email=m.get("officeEmail"),
|
|
)
|
|
for m in data
|
|
]
|
|
|
|
psa_cache.set(cache_key, result, ttl_seconds=900)
|
|
return result
|
|
|
|
# ── Ticket Context ────────────────────────────────────────────────
|
|
|
|
async def get_ticket_context(
|
|
self, ticket_id: int, connection_id: str | None = None
|
|
):
|
|
"""Fetch rich ticket context for AI prompt injection.
|
|
|
|
Returns a TicketContext with ticket details, company, contact,
|
|
configurations, recent notes, and related open tickets.
|
|
Results are cached for 5 minutes per ticket.
|
|
"""
|
|
from app.schemas.psa_context import (
|
|
TicketContext,
|
|
TicketDetails,
|
|
CompanyInfo,
|
|
ContactInfo,
|
|
ConfigItem,
|
|
TicketNote,
|
|
RelatedTicket,
|
|
)
|
|
|
|
cache_key = f"{connection_id or 'default'}:ticket_context:{ticket_id}"
|
|
cached = psa_cache.get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
# Fetch ticket first to get company_id and contact_id
|
|
ticket_data = await self.client.get(
|
|
f"/service/tickets/{ticket_id}",
|
|
params={
|
|
"fields": "id,summary,status,priority,board,sla,dateEntered,resources,company,contact"
|
|
},
|
|
)
|
|
|
|
company_id = ticket_data.get("company", {}).get("id") if ticket_data.get("company") else None
|
|
contact_id = ticket_data.get("contact", {}).get("id") if ticket_data.get("contact") else None
|
|
|
|
# Build parallel fetch tasks
|
|
configs_task = asyncio.create_task(
|
|
self.client.get(
|
|
f"/service/tickets/{ticket_id}/configurations",
|
|
params={
|
|
"fields": "id,deviceIdentifier,type,osType,serialNumber,ipAddress,modelNumber"
|
|
},
|
|
)
|
|
)
|
|
notes_task = asyncio.create_task(
|
|
self.client.get(
|
|
f"/service/tickets/{ticket_id}/notes",
|
|
params={
|
|
"pageSize": "20",
|
|
"orderBy": "dateCreated desc",
|
|
"fields": "id,text,member,dateCreated,internalAnalysisFlag",
|
|
},
|
|
)
|
|
)
|
|
company_task = asyncio.create_task(
|
|
self.client.get(
|
|
f"/company/companies/{company_id}",
|
|
params={
|
|
"fields": "id,name,site,addressLine1,city,state,zip,phoneNumber,type,territory"
|
|
},
|
|
)
|
|
) if company_id else None
|
|
|
|
related_task = asyncio.create_task(
|
|
self.client.get(
|
|
"/service/tickets",
|
|
params={
|
|
"conditions": f"company/id={company_id} AND closedFlag=false AND id != {ticket_id}",
|
|
"pageSize": "5",
|
|
"orderBy": "id desc",
|
|
"fields": "id,summary,status,priority,board",
|
|
},
|
|
)
|
|
) if company_id else None
|
|
|
|
contact_task = asyncio.create_task(
|
|
self.client.get(
|
|
f"/company/contacts/{contact_id}",
|
|
params={
|
|
"fields": "id,firstName,lastName,title,defaultPhoneNbr,communicationItems"
|
|
},
|
|
)
|
|
) if contact_id else None
|
|
|
|
# Gather all tasks with partial failure tolerance
|
|
tasks_to_await = [t for t in [configs_task, notes_task, company_task, related_task, contact_task] if t is not None]
|
|
task_results = await asyncio.gather(*tasks_to_await, return_exceptions=True)
|
|
|
|
# Unpack results in order (skipping None tasks)
|
|
result_iter = iter(task_results)
|
|
configs_raw = next(result_iter)
|
|
notes_raw = next(result_iter)
|
|
company_raw = next(result_iter) if company_task else None
|
|
related_raw = next(result_iter) if related_task else None
|
|
contact_raw = next(result_iter) if contact_task else None
|
|
|
|
# Map ticket details
|
|
def _parse_dt(val: str | None) -> datetime:
|
|
if not val:
|
|
return datetime.now(timezone.utc)
|
|
try:
|
|
# CW returns ISO 8601 strings — ensure timezone aware
|
|
dt = datetime.fromisoformat(val.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
except (ValueError, AttributeError):
|
|
return datetime.now(timezone.utc)
|
|
|
|
ticket_details = TicketDetails(
|
|
id=ticket_data["id"],
|
|
summary=ticket_data.get("summary", ""),
|
|
status=ticket_data.get("status", {}).get("name", "") if isinstance(ticket_data.get("status"), dict) else str(ticket_data.get("status", "")),
|
|
priority=ticket_data.get("priority", {}).get("name", "") if isinstance(ticket_data.get("priority"), dict) else str(ticket_data.get("priority", "")),
|
|
board=ticket_data.get("board", {}).get("name", "") if isinstance(ticket_data.get("board"), dict) else str(ticket_data.get("board", "")),
|
|
sla=ticket_data.get("sla", {}).get("name") if isinstance(ticket_data.get("sla"), dict) else ticket_data.get("sla"),
|
|
date_entered=_parse_dt(ticket_data.get("dateEntered")),
|
|
resources=ticket_data.get("resources"),
|
|
)
|
|
|
|
# Map company
|
|
company_info: CompanyInfo
|
|
if isinstance(company_raw, dict):
|
|
addr_parts = [
|
|
company_raw.get("addressLine1"),
|
|
company_raw.get("city"),
|
|
company_raw.get("state"),
|
|
company_raw.get("zip"),
|
|
]
|
|
address = ", ".join(p for p in addr_parts if p) or None
|
|
company_info = CompanyInfo(
|
|
id=company_raw["id"],
|
|
name=company_raw.get("name", ""),
|
|
site=company_raw.get("site", {}).get("name") if isinstance(company_raw.get("site"), dict) else company_raw.get("site"),
|
|
address=address,
|
|
phone=company_raw.get("phoneNumber"),
|
|
type=company_raw.get("type", {}).get("name") if isinstance(company_raw.get("type"), dict) else company_raw.get("type"),
|
|
territory=company_raw.get("territory", {}).get("name") if isinstance(company_raw.get("territory"), dict) else company_raw.get("territory"),
|
|
)
|
|
else:
|
|
if isinstance(company_raw, Exception):
|
|
logger.warning("Failed to fetch company for ticket %s: %s", ticket_id, company_raw)
|
|
# Fallback: use data from ticket itself
|
|
company_info = CompanyInfo(
|
|
id=company_id or 0,
|
|
name=ticket_data.get("company", {}).get("name", "") if isinstance(ticket_data.get("company"), dict) else "",
|
|
)
|
|
|
|
# Map contact
|
|
contact_info: ContactInfo | None = None
|
|
if isinstance(contact_raw, dict):
|
|
first = contact_raw.get("firstName", "")
|
|
last = contact_raw.get("lastName", "")
|
|
full_name = f"{first} {last}".strip() or "Unknown"
|
|
|
|
# Extract email from communicationItems
|
|
email: str | None = None
|
|
comm_items = contact_raw.get("communicationItems", [])
|
|
if isinstance(comm_items, list):
|
|
for item in comm_items:
|
|
if isinstance(item, dict) and item.get("communicationType") == "Email":
|
|
email = item.get("value")
|
|
break
|
|
|
|
contact_info = ContactInfo(
|
|
name=full_name,
|
|
email=email,
|
|
phone=contact_raw.get("defaultPhoneNbr"),
|
|
title=contact_raw.get("title"),
|
|
)
|
|
elif isinstance(contact_raw, Exception):
|
|
logger.warning("Failed to fetch contact for ticket %s: %s", ticket_id, contact_raw)
|
|
|
|
# Map configurations
|
|
configurations: list[ConfigItem] = []
|
|
if isinstance(configs_raw, list):
|
|
for cfg in configs_raw:
|
|
if not isinstance(cfg, dict):
|
|
continue
|
|
configurations.append(ConfigItem(
|
|
device_identifier=cfg.get("deviceIdentifier", ""),
|
|
type=cfg.get("type", {}).get("name") if isinstance(cfg.get("type"), dict) else cfg.get("type"),
|
|
os_type=cfg.get("osType", {}).get("name") if isinstance(cfg.get("osType"), dict) else cfg.get("osType"),
|
|
serial_number=cfg.get("serialNumber"),
|
|
ip_address=cfg.get("ipAddress"),
|
|
model_number=cfg.get("modelNumber"),
|
|
))
|
|
elif isinstance(configs_raw, Exception):
|
|
logger.warning("Failed to fetch configs for ticket %s: %s", ticket_id, configs_raw)
|
|
|
|
# Map notes
|
|
notes: list[TicketNote] = []
|
|
if isinstance(notes_raw, list):
|
|
for note in notes_raw:
|
|
if not isinstance(note, dict):
|
|
continue
|
|
member_name: str | None = None
|
|
member_obj = note.get("member")
|
|
if isinstance(member_obj, dict):
|
|
first = member_obj.get("firstName", "")
|
|
last = member_obj.get("lastName", "")
|
|
member_name = f"{first} {last}".strip() or member_obj.get("identifier")
|
|
elif isinstance(member_obj, str):
|
|
member_name = member_obj
|
|
|
|
notes.append(TicketNote(
|
|
text=note.get("text", ""),
|
|
member=member_name,
|
|
date_created=_parse_dt(note.get("dateCreated")),
|
|
internal_analysis_flag=note.get("internalAnalysisFlag", False),
|
|
))
|
|
elif isinstance(notes_raw, Exception):
|
|
logger.warning("Failed to fetch notes for ticket %s: %s", ticket_id, notes_raw)
|
|
|
|
# Map related tickets
|
|
related_tickets: list[RelatedTicket] = []
|
|
if isinstance(related_raw, list):
|
|
for rt in related_raw:
|
|
if not isinstance(rt, dict):
|
|
continue
|
|
related_tickets.append(RelatedTicket(
|
|
id=rt["id"],
|
|
summary=rt.get("summary", ""),
|
|
status=rt.get("status", {}).get("name", "") if isinstance(rt.get("status"), dict) else str(rt.get("status", "")),
|
|
priority=rt.get("priority", {}).get("name", "") if isinstance(rt.get("priority"), dict) else str(rt.get("priority", "")),
|
|
board=rt.get("board", {}).get("name", "") if isinstance(rt.get("board"), dict) else str(rt.get("board", "")),
|
|
))
|
|
elif isinstance(related_raw, Exception):
|
|
logger.warning("Failed to fetch related tickets for ticket %s: %s", ticket_id, related_raw)
|
|
|
|
ctx = TicketContext(
|
|
ticket=ticket_details,
|
|
company=company_info,
|
|
contact=contact_info,
|
|
configurations=configurations,
|
|
notes=notes,
|
|
related_tickets=related_tickets,
|
|
fetched_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
psa_cache.set(cache_key, ctx, ttl_seconds=300)
|
|
return ctx
|
|
|
|
async def create_time_entry(
|
|
self,
|
|
ticket_id: str,
|
|
member_id: str,
|
|
hours: float,
|
|
notes: str | None = None,
|
|
work_type: str | None = None,
|
|
) -> PSATimeEntry:
|
|
"""Create a time entry on a CW ticket via POST /time/entries."""
|
|
payload: dict = {
|
|
"chargeToId": int(ticket_id),
|
|
"chargeToType": "ServiceTicket",
|
|
"member": {"id": int(member_id)},
|
|
"timeStart": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"actualHours": hours,
|
|
}
|
|
if notes:
|
|
payload["notes"] = notes[:2000] # CW limit
|
|
if work_type:
|
|
payload["workType"] = {"name": work_type}
|
|
|
|
data = await self.client.post("/time/entries", payload)
|
|
return PSATimeEntry(
|
|
id=str(data["id"]),
|
|
ticket_id=ticket_id,
|
|
member_id=member_id,
|
|
hours=data.get("actualHours", hours),
|
|
notes=data.get("notes"),
|
|
created_at=data.get("timeStart"),
|
|
)
|
|
|
|
# ── Private helpers ───────────────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _map_ticket(data: dict) -> PSATicket:
|
|
"""Map a CW ticket JSON dict to a PSATicket."""
|
|
return PSATicket(
|
|
id=str(data["id"]),
|
|
summary=data.get("summary", ""),
|
|
company_name=data.get("company", {}).get("name"),
|
|
company_id=str(data["company"]["id"]) if data.get("company") else None,
|
|
board_name=data.get("board", {}).get("name"),
|
|
board_id=data.get("board", {}).get("id"),
|
|
status_name=data.get("status", {}).get("name"),
|
|
status_id=data.get("status", {}).get("id"),
|
|
priority_name=data.get("priority", {}).get("name"),
|
|
priority_id=data.get("priority", {}).get("id"),
|
|
closed=data.get("closedFlag", False),
|
|
)
|