Backend patterns

This page is the pattern library for the Lockstep VM backend. Each pattern is a named, canonical skeleton that already appears two-or-more times in the codebase — copy it verbatim and adapt the vendor-specific parts. Everything here is verified against webapp/backend/app.

The five patterns compose into one pipeline:

Integration / CSV Provider  →  inventory_items (source=<key>)  →  Rollup  →  applications
                            →  <source>_devices / csv_devices   ┐
                            →  entra_users / csv_users           ├→  Identity reconciliation  →  Assets / Users surfaces

Pattern	When to reach for it	Canonical files
Integration	Vendor has an API you poll	`endpoints/{intune,connectwise_automate,entra,sentinelone,rapid7}.py`
CSV Provider	Source is a manual CSV/XLSX export, no API	`services/csv_providers/*`
Rollup	Any source that writes software rows	`services/inventory_rollup.py`
Identity reconciliation	Same machine/person seen under different keys	`endpoints/devices.py`, `endpoints/entra.py`
Env-seeded default config	Credentials live in env, want zero-touch setup	`main.py` `_seed_*`

1. Integration pattern

An integration polls a vendor API on demand, upserts the results into MongoDB scoped by tenant_id, prunes rows the vendor no longer reports, rebuilds the software rollup, and records sync status on a per-tenant config document. intune.py, connectwise_automate.py, entra.py, sentinelone.py and rapid7.py are all instances of this pattern; they differ only in the client, the field mapping and the collections written.

Every integration module has the same seven parts, in this order.

1a. Config Pydantic model

Credentials plus an enabled flag. Secrets are Optional so the UI can PATCH the config without re-sending an unchanged secret.

class IntuneConfig(BaseModel):
    azure_tenant_id: str
    azure_client_id: str
    auth_method: str = "secret"            # "secret" | "certificate"
    client_secret: Optional[str] = None
    cert_thumbprint: Optional[str] = None
    cert_private_key: Optional[str] = None  # PEM text
    enabled: bool = True

1b. Config collection + conventions

Each integration owns a <vendor>_config collection with one document per tenant_id. The document carries both credentials and mutable sync state. The standard state fields (write them from _run_sync, read them from status routes):

Field	Meaning
`tenant_id`	Partition key — every query filters on it
`enabled`	Integration toggle
`sync_status`	`null` \| `"running"` \| `"success"` \| `"error"`
`sync_started_at`, `last_sync`	Timestamps
`sync_error`	Last error string, or `null` on success
`device_count` / `app_count` / `user_count` …	Last-sync tallies for the UI
`updated_at`	Last config write

1c. `_build_client` — construct the vendor client from a config doc

Pure function; no DB access. Keeps client construction in one place so routes and the sync job build it identically.

def _build_client(config_doc: dict) -> IntuneClient:
    return IntuneClient(
        azure_tenant_id=config_doc["azure_tenant_id"],
        azure_client_id=config_doc["azure_client_id"],
        client_secret=config_doc.get("client_secret"),
        cert_thumbprint=config_doc.get("cert_thumbprint"),
        cert_private_key=config_doc.get("cert_private_key"),
    )

1d. `_resolve_config` — adopt-on-read

The linchpin of multi-tenant credential sharing. A tenant with no config of its own adopts the env-seeded "default" config on first read, via $setOnInsert so a concurrent adopt can’t double-write. After adoption the tenant has its own copy, so per-tenant edits/deletes win afterwards. Pair this with the env-seed that creates the default doc.

async def _resolve_config(tenant_id: str, db: AsyncIOMotorDatabase) -> Optional[dict]:
    config = await db.intune_config.find_one({"tenant_id": tenant_id})
    if config:
        return config
    if tenant_id != "default":
        seed = await db.intune_config.find_one({"tenant_id": "default"})
        if seed:
            adopted = {k: v for k, v in seed.items() if k != "_id"}
            adopted["tenant_id"] = tenant_id
            await db.intune_config.update_one(
                {"tenant_id": tenant_id}, {"$setOnInsert": adopted}, upsert=True
            )
            return await db.intune_config.find_one({"tenant_id": tenant_id})
    return None


async def _get_config_or_404(tenant_id: str, db: AsyncIOMotorDatabase) -> dict:
    config = await _resolve_config(tenant_id, db)
    if not config:
        raise HTTPException(status_code=404, detail="Intune not configured. POST /intune/config first.")
    return config

Always route reads through _resolve_config (or _get_config_or_404) — never find_one the config collection directly from a route, or adoption won’t happen. A _safe_config(doc) helper strips secrets and returns has_secret/has_cert booleans for GET responses.

1e. `_run_sync` — the background sync job

The heart of the pattern. Signature is always (tenant_id, db) so it can be handed to BackgroundTasks.add_task. Its shape is invariant: mark running → fetch (in an executor) → bulk-upsert → prune stale → write inventory_items → rebuild rollup → mark success, wrapped in try/except that records sync_status="error".

Key rules, all load-bearing:

Blocking HTTP runs in a thread-pool executor (await loop.run_in_executor(None, fn)). The vendor clients are synchronous requests-based; running them inline would block the event loop and stall auth/login for every tenant during a sync.
Bulk writes use pymongo.UpdateOne(..., upsert=True) collected into a list and flushed with bulk_write(ops, ordered=False). Chunk at 1000 ops for large sets.
Upsert key is {tenant_id, <vendor id field>} for entity collections, and {tenant_id, source, device_id, application_name, version} for inventory_items.
Prune stale: collect the ids seen this run into live_* sets and delete_many({... , <id>: {"$nin": list(live_ids)}}). This is how deletions on the vendor side propagate.
Rebuild the rollup at the end, non-fatally (see Rollup).

async def _run_sync(tenant_id: str, db: AsyncIOMotorDatabase) -> None:
    import asyncio
    from pymongo import UpdateOne

    try:
        await db.intune_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "running", "sync_started_at": datetime.utcnow(), "sync_error": None}},
        )
        config_doc = await db.intune_config.find_one({"tenant_id": tenant_id})
        client = _build_client(config_doc)

        # 1. Fetch — ALL blocking HTTP in the executor, never on the event loop.
        loop = asyncio.get_event_loop()
        def _fetch_all():
            return client.get_devices(), client.get_apps()
        devices, apps = await loop.run_in_executor(None, _fetch_all)

        now = datetime.utcnow()

        # 2. Bulk-upsert entities + track live ids.
        live_device_ids: set[str] = set()
        dev_ops = []
        for d in devices:
            did = d.get("intune_device_id")
            if did:
                live_device_ids.add(did)
            dev_ops.append(UpdateOne(
                {"tenant_id": tenant_id, "intune_device_id": did},
                {"$set": {**d, "tenant_id": tenant_id, "synced_at": now}},
                upsert=True,
            ))
        if dev_ops:
            await db.intune_devices.bulk_write(dev_ops, ordered=False)

        # 3. Prune stale (rows the vendor no longer reports).
        if live_device_ids:
            await db.intune_devices.delete_many(
                {"tenant_id": tenant_id, "intune_device_id": {"$nin": list(live_device_ids)}}
            )

        # 4. Write software rows into the shared inventory_items collection.
        inv_ops = []
        for d in devices:
            for app in d.get("apps", []):
                app_name = (app.get("name") or "").strip()
                if not app_name:
                    continue
                inv_ops.append(UpdateOne(
                    {"tenant_id": tenant_id, "source": "intune",
                     "device_id": d["intune_device_id"],
                     "application_name": app_name, "version": app.get("version", "")},
                    {"$set": {
                        "tenant_id": tenant_id, "source": "intune",
                        "device_id": d["intune_device_id"],
                        "device_name": d.get("device_name", ""),
                        "application_name": app_name,
                        "vendor": app.get("publisher", ""),
                        "version": app.get("version", ""),
                        "os_platform": d.get("os", ""),
                        "user_email": d.get("user_upn", ""),
                        "user_name": d.get("user_display_name", ""),
                        "synced_at": now,
                    }},
                    upsert=True,
                ))
        for i in range(0, len(inv_ops), 1000):
            await db.inventory_items.bulk_write(inv_ops[i:i + 1000], ordered=False)
        if live_device_ids:
            await db.inventory_items.delete_many(
                {"tenant_id": tenant_id, "source": "intune",
                 "device_id": {"$nin": list(live_device_ids)}}
            )

        # 5. Rebuild the Software-inventory rollup (non-fatal).
        try:
            await rebuild_applications(tenant_id, db)
        except Exception as rollup_exc:
            logger.warning("Intune applications rollup failed (non-fatal): %s", rollup_exc)

        # 6. Mark success + tallies.
        await db.intune_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "success", "last_sync": now, "sync_error": None,
                      "device_count": len(devices), "app_count": len(apps)}},
        )
    except (IntuneAuthError, IntuneAPIError) as exc:
        await db.intune_config.update_one(
            {"tenant_id": tenant_id}, {"$set": {"sync_status": "error", "sync_error": str(exc)}})
    except Exception as exc:
        logger.exception("Unexpected Intune sync error for tenant=%s", tenant_id)
        await db.intune_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "error", "sync_error": f"Unexpected error: {exc}"}})

Circuit breaker (optional). When a per-entity sub-resource can fail independently (e.g. ConnectWise Automate’s /Software endpoint), guard the loop with a consecutive-failure counter and stop calling it after a limit instead of hammering every device. See connectwise_automate.py _run_sync (SW_FAIL_LIMIT = 15).

1f. The standard routes

Every integration exposes the same nine routes off its own prefix (registered in app/api/v1/__init__.py, e.g. prefix="/intune"). All depend on get_current_user and get_database.

Route	Purpose
`POST /config`	Save credentials. Only overwrite a secret when a new value is supplied; validate `auth_method` requirements.
`GET /config`	`{"configured": bool, **_safe_config}` — secrets redacted.
`DELETE /config`	Delete the config and all synced data for the tenant.
`POST /test-connection`	Live credential check; maps auth errors → 401, API errors → 502.
`POST /sync`	Guard on `sync_status == "running"`, else `background_tasks.add_task(_run_sync, tenant_id, db)`.
`GET /sync-status`	Current `sync_status` / `last_sync` / counts.
`GET /stats`	Aggregated summary for the dashboard.
`GET /devices` (or `/apps`, `/users`)	Paginated list of synced rows.
Extra vendor routes	e.g. Automate’s `GET /clients` (scoping), `GET /device-by-name/{name}`.

Canonical trigger + config-save routes:

@router.post("/sync")
async def trigger_sync(
    background_tasks: BackgroundTasks,
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    config_doc = await _get_config_or_404(current_user["tenant_id"], db)
    if config_doc.get("sync_status") == "running":
        return {"message": "Sync already in progress", "sync_status": "running"}
    background_tasks.add_task(_run_sync, current_user["tenant_id"], db)
    return {"message": "Sync started", "sync_status": "running"}


@router.post("/config")
async def configure(
    config: IntuneConfig,
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    tenant_id = current_user["tenant_id"]
    doc = {
        "tenant_id": tenant_id,
        "azure_tenant_id": config.azure_tenant_id,
        "azure_client_id": config.azure_client_id,
        "auth_method": config.auth_method,
        "enabled": config.enabled,
        "updated_at": datetime.utcnow(),
    }
    # Only overwrite secrets when supplied — partial saves must not wipe them.
    if config.client_secret:
        doc["client_secret"] = config.client_secret
    if config.cert_thumbprint:
        doc["cert_thumbprint"] = config.cert_thumbprint
    if config.cert_private_key:
        doc["cert_private_key"] = config.cert_private_key
    await db.intune_config.update_one({"tenant_id": tenant_id}, {"$set": doc}, upsert=True)
    return {"configured": True, "message": "Intune credentials saved"}

2. CSV Provider pattern

For sources with no API — data arrives as a manual CSV/XLSX export. The CSV Provider framework (services/csv_providers/) mirrors the integration lifecycle (client→config→sync→data→rollup) where the “client” is a Python parser and the “sync” is an upload+parse job. One generic set of routes in endpoints/csv_sources.py drives every provider off the registry — you never write endpoints for a new CSV source.

2a. Write a `CsvProvider` subclass

Set the descriptor class attributes and implement parse(rows, tenant_id) -> ParseResult. The base class (csv_providers/base.py) already handles decoding (CSV vs XLSX by magic bytes), preamble skipping (header_signals), and loose column matching (pick, pick_contains, has_columns). Do not override load() unless the file shape is unusual.

from .base import CsvProvider, ParseResult, normalize_platform

class SoftwareInventoryProvider(CsvProvider):
    key = "software_inventory"          # unique; also the inventory_items `source` value
    name = "Software Inventory CSV"
    description = "Generic installed-software export: one row per device + application."
    category = "software"               # software | identity | access (UI grouping)
    target = "software"                 # software | device | user | records (persist path)
    icon = "Package"                    # lucide icon name for the UI card
    accepts = "CSV with device, application, and version columns"
    expected_columns = ["Device / Hostname", "Application / Software", "Version", "Vendor", "OS"]
    # header_signals = ["Person Name", "Card No"]   # set only if the export has a preamble

    def parse(self, rows: list[dict], tenant_id: str) -> ParseResult:
        records, warnings = [], []
        for row in rows:
            app = self.pick(row, "application_name", "application", "software", "name", "product")
            device = self.pick(row, "device_name", "device", "hostname", "computer", "machine")
            if not app:
                continue
            records.append({
                "device_id": self.pick(row, "device_id", "asset_id", "hostname") or device,
                "device_name": device,
                "application_name": app,
                "vendor": self.pick(row, "vendor", "publisher", "manufacturer"),
                "version": self.pick(row, "version", "app_version", "product_version"),
                "os_platform": normalize_platform(self.pick(row, "os", "operating_system", "platform")),
                "user_name": self.pick(row, "user", "user_name", "assigned_user"),
            })
        if not records:
            warnings.append("No rows had a recognizable application/name column.")
        return ParseResult(records=records, warnings=warnings, detected_shape="software")

self.pick(row, *names) returns the first non-empty value among candidate column names, matched case/space/punctuation-insensitively (_norm). self.pick_contains(row, *substrings) matches columns whose header contains a substring (for noisy headers like "Supervisor's Name (First Last)").

2b. The `target` decides the persist path

target is the single most important attribute: it selects which collection the generic upload job writes to and the reconciliation key. Choose it by what the data is, not where it came from.

`target`	Collection	Upsert key	Rollup?	Use for
`software`	`inventory_items` (`source=<key>`)	`{tenant_id, source, device_id, application_name, version}`	✅ `rebuild_applications`	Installed-software lists
`device`	`csv_devices`	`{tenant_id, source, device_key}` (`device_key = name.lower()`)	✅ links by hostname	Endpoint posture/enrichment (e.g. AutoElevate). Never creates assets.
`user`	`csv_users`	`{tenant_id, source, email}` (lowercased)	✅ links by email	Badge rosters, HR — enrich Entra users
`records`	`csv_records`	replace-all per source	—	Reference data with no join target

The device and user targets are enrichment, not ingestion: they key on hostname / email so the row links to an existing asset or user and shows up as a tab/attributes there — they never mint duplicate assets or software. software, device, user are delete-and-replace or prune per source on each upload (a fresh export is the authoritative full set).

The record shapes each target expects from parse():

software → {device_id, device_name, application_name, vendor, version, os_platform, user_name}
device → {device_name, attributes: {...}}
user → {email, display_name, site, attributes: {...}, flags: {...}}
records → any flat dict (stored as-is)

2c. Register the provider

Append an instance to _PROVIDERS in csv_providers/__init__.py. That is the only wiring step — endpoints, upload jobs, UI cards and detail pages are all driven off the registry generically.

_PROVIDERS: List[CsvProvider] = [
    SoftwareInventoryProvider(),
    AutoElevateProvider(),     # target="device"  — links by hostname
    BadgeAccessProvider(),     # target="user"    — links by email, per-site
    HrRosterProvider(),        # target="user"
    MyNewProvider(),           # ← add here
]

REGISTRY: Dict[str, CsvProvider] = {p.key: p for p in _PROVIDERS}

See Extending recipes for the full add-a-source walkthrough and Data sources for the user-facing view.

3. Rollup pattern

inventory_items holds one row per (device, application, version, source) and is the shared source of truth every source writes to. The Software-inventory UI, however, reads applications: a per-tenant rollup grouped by software with a distinct-device count, version tracking and source attribution. rebuild_applications(tenant_id, db) in services/inventory_rollup.py rebuilds that rollup from inventory_items — call it at the end of every sync/upload that touches software rows.

from app.services.inventory_rollup import rebuild_applications

try:
    await rebuild_applications(tenant_id, db)
except Exception as rollup_exc:
    logger.warning("applications rollup failed (non-fatal): %s", rollup_exc)

Always call it non-fatally — a rollup failure must not fail the sync that already wrote good data.

What `rebuild_applications` does

Aggregate inventory_items for the tenant, grouping by application_name (name only — a single app can carry multiple vendor spellings; the dominant vendor, seen on the most distinct devices, is chosen for display). Collects distinct (version, device) and (vendor, device) pairs, the set of os_platforms, and the set of sources.
Normalize the name (_normalize_app_name) and merge raw-name groups that collapse to the same normalized name, unioning their device/version/source sets. Normalization strips a trailing per-install instance id / GUID in parentheses so every install collapses into one group, while deliberately keeping meaningful parentheticals:
- ScreenConnect Client (51a7836ecd7be670) → ScreenConnect Client
- 7-Zip 22.01 (x64) → unchanged (x64 is not a hex id)
Compute per group: distinct device_count, per-version device counts (version_tracking — most-common / newest / oldest, ordered by a numeric-aware _version_key), deduped os_platforms, and the sorted sources list. Version counts are stored as a list of {version, count}, never a dict keyed by version, so arbitrary version strings can’t produce an invalid MongoDB field name.
Source tagging: sources = sorted({s for s in group.sources if s}) — this array is what drives the source chips (intune, sentinelone, rapid7, connectwise_automate, or any CSV provider key) on each software row. It is why every writer must set a distinct source on its inventory_items rows.
Upsert each group into applications keyed by {tenant_id, normalized_name}, stamping rollup_at = run_ts.
Prune stale: delete_many({tenant_id, rollup_at: {"$ne": run_ts}}) drops rollup docs no longer backed by any inventory row.

The upshot for an integration author: you never write to applications directly. Write clean inventory_items rows with a correct source, then call rebuild_applications. Field mapping and hygiene: use _clean semantics (strip NULL/control chars) on names and versions, since MongoDB rejects control chars in field names and some agent exports include them.

See the Data model for the full inventory_items and applications shapes.

4. Identity reconciliation pattern

The same physical machine and the same person show up under different keys across sources. Two reconciliation techniques stitch them into one asset / one user. Reconciliation happens at read time in the query pipeline — the raw rows keep their source-native keys; nothing is mutated on write.

4a. Device reconciliation via the NetBIOS key

ConnectWise Automate reports the 15-char NetBIOS hostname (uppercased, e.g. LTG-ARAMARAJU-L), while Intune / SentinelOne report the full DNS name (LTG-ARamaraju-LT), and Rapid7 may report an FQDN. The canonical device key is the uppercased first-15-characters of the hostname. endpoints/devices.py POST /devices/search groups inventory_items on that key, then picks the longest observed name as the display name and $lookups each source collection by lowercased name.

base_stages = [
    {"$match": pre_filter},                       # {tenant_id, ...}
    {"$sort": {"source": 1, "device_name": 1}},
    # Device-identity key: uppercased NetBIOS-truncated (15-char) hostname.
    {"$addFields": {"_nbkey": {"$toUpper": {"$substrCP": [
        {"$ifNull": ["$device_name", ""]}, 0, 15]}}}},
    {"$group": {
        "_id": "$_nbkey",
        "_names":      {"$addToSet": "$device_name"},
        "device_id":   {"$first": "$device_id"},
        "sources":     {"$addToSet": {"$ifNull": ["$source", "intune"]}},
        "application_count": {"$sum": 1},
        "applications":      {"$addToSet": "$application_name"},
    }},
    # Canonical display name = the longest observed name (the full DNS form).
    {"$addFields": {"device_name": {"$reduce": {
        "input": "$_names", "initialValue": "",
        "in": {"$cond": [{"$gt": [{"$strLenCP": "$$this"}, {"$strLenCP": "$$value"}]},
                         "$$this", "$$value"]}}}}},
    {"$addFields": {"_name_lc": {"$toLower": "$device_name"}}},
    # Per-source $lookup by lowercased name; drop `_id` (ObjectId isn't JSON-safe).
    {"$lookup": {
        "from": "intune_devices",
        "let": {"n": "$_name_lc"},
        "pipeline": [
            {"$match": {"tenant_id": tenant_id}},
            {"$match": {"$expr": {"$eq": [{"$toLower": "$device_name"}, "$$n"]}}},
            {"$project": {"_id": 0}}, {"$limit": 1},
        ],
        "as": "intune",
    }},
    {"$addFields": {"intune": {"$arrayElemAt": ["$intune", 0]}}},
    # Prefer the Intune UUID for the asset-detail deep link.
    {"$addFields": {"device_id": {"$ifNull": ["$intune.intune_device_id", "$device_id"]}}},
    # …repeat the $lookup block for sentinelone_devices, rapid7_assets…
]

Conventions that fall out of this:

device_id preference order for the asset-detail deep link: Intune UUID (intune_device_id) → the device_id already on the inventory row → the name itself. The simpler GET /devices list applies the same preference procedurally.
Rapid7 hostnames may be FQDNs — match on both the full lowercased hostname and its short form ($split on ., take element 0).
CSV device sources (csv_devices, keyed by device_key) are matched on both the exact lowercased name and the 15-char NetBIOS form, so an AutoElevate row links regardless of which form the export used.
Single-record lookups (e.g. connectwise_automate.py GET /device-by-name/{name}) match $or on the exact name and the name.upper()[:15] NetBIOS form.

4b. User reconciliation via Entra + SMTP aliases

A person’s badge/HR row may use an email that is an alias, not their Entra UPN. endpoints/entra.py _run_sync extracts every SMTP alias from proxyAddresses (the smtp:/SMTP:-prefixed entries) into an aliases array on the entra_users doc:

aliases = []
for pa in (u.get("proxyAddresses") or []):
    if isinstance(pa, str) and pa.lower().startswith("smtp:"):
        a = pa.split(":", 1)[1].strip().lower()
        if a:
            aliases.append(a)
# → stored as: "aliases": sorted(set(aliases))

GET /entra/user/{upn} then links enrichment (csv_users from badge/HR providers) by matching against the union of UPN, primary mail, and every alias:

email_set  = {(user.get("upn") or "").lower(), (user.get("mail") or "").lower()}
email_set |= {a.lower() for a in (user.get("aliases") or [])}
emails = [e for e in email_set if e]
enrichment = []
if emails:
    async for cu in db.csv_users.find({"tenant_id": tid, "email": {"$in": emails}}, {"_id": 0}):
        enrichment.append({"source": cu.get("source"), "site": cu.get("site", ""),
                           "attributes": cu.get("attributes", {}), "flags": cu.get("flags", {})})
user["enrichment"] = enrichment
user["badge_admin"] = any(e.get("flags", {}).get("badge_admin") for e in enrichment)

The same endpoint links a user to their assets via _linked_devices(tenant_id, upn, display_name, db): Intune devices carry the owner UPN; inventory_items carry user_email / user_name. It collects matching device names (case-insensitive exact match), then resolves each name back to the device_id the asset-detail page accepts — again preferring the Intune UUID.

Takeaway: case-insensitive matching everywhere ({"$regex": f"^{re.escape(value)}$", "$options": "i"}); reconcile by name/email at read time; and always resolve to the Intune UUID when producing a deep link. See Users and Assets for the surfaces this powers.

5. Env-seeded default config pattern

To give a fresh deployment zero-touch setup, credentials supplied via environment variables are seeded into the "default" tenant’s config on startup. Combined with adopt-on-read, env credentials then apply to every tenant automatically, while any tenant’s manual edit/delete still wins.

Each integration has an async def _seed_<vendor>_config() in app/main.py, all awaited from the FastAPI lifespan startup after the Mongo connection opens:

@asynccontextmanager
async def lifespan(app: FastAPI):
    await connect_to_mongo()
    await _seed_intune_config()
    await _seed_sentinelone_config()
    await _seed_connectwise_config()
    await _seed_rapid7_config()
    await _seed_connectwise_automate_config()
    await _seed_entra_config()
    yield
    await close_mongo_connection()

The seed function is idempotent: it returns early if the required env vars are absent, and returns early if a default config already exists — so restarts and manual overrides are never clobbered.

async def _seed_connectwise_automate_config() -> None:
    """Seed the default-tenant config from env vars if not already set, so it has
    the same adopt-on-read lifecycle as every other integration."""
    if not (settings.CW_AUTOMATE_URL and settings.CW_AUTOMATE_USERNAME and settings.CW_AUTOMATE_PASSWORD):
        return
    try:
        db = get_database()
        existing = await db.connectwise_automate_config.find_one({"tenant_id": "default"})
        if existing:
            logger.info("ConnectWise Automate config already exists for default tenant — skipping seed")
            return
        await db.connectwise_automate_config.insert_one({
            "tenant_id": "default",
            "api_url": settings.CW_AUTOMATE_URL,
            "username": settings.CW_AUTOMATE_USERNAME,
            "password": settings.CW_AUTOMATE_PASSWORD,
            "client_id": settings.CW_AUTOMATE_CLIENT_ID,
            "enabled": True,
            "sync_status": None,
            "last_sync": None,
            "updated_at": datetime.utcnow(),
        })
        logger.info("ConnectWise Automate config seeded from env vars (url=%s)", settings.CW_AUTOMATE_URL)
    except Exception as exc:
        logger.error("Failed to seed ConnectWise Automate config: %s", exc)

Checklist for a new integration’s seed:

Add the settings fields to app/core/config.py (Settings reads them from env — e.g. CW_AUTOMATE_URL: str = "").
Write _seed_<vendor>_config() following the guard → dedupe → insert_one({"tenant_id": "default", ...}) shape. Seed the same field names your _build_client reads and the same state fields (sync_status=None, last_sync=None, updated_at).
Await it in lifespan, wrapped so a failure only logs (never blocks startup).

Notes from the existing seeds:

Certificate auth (_seed_intune_config, _seed_entra_config) reads a PEM from INTUNE_CERT_KEY_PATH, calls _extract_private_key, and skips the seed with a warning if the file is missing.
Reusing an app registration: _seed_entra_config seeds entra_config from the same Intune Azure app + cert (Entra just needs User.Read.All / Group.Read.All consented) — a source can piggyback on another’s credentials.
Some seeds delegate env parsing to the endpoint module (_seed_connectwise_config imports _cw_config_from_env lazily to avoid a circular import at module load).

Copy-this checklist for a new API integration

Client — a synchronous services/<vendor>_client.py with typed *AuthError / *APIError exceptions.
Endpoint module endpoints/<vendor>.py — Config model, _build_client, _resolve_config + _get_config_or_404, _safe_config, _run_sync, and the nine standard routes.
Register the router in app/api/v1/__init__.py with prefix="/<vendor>".
Write source="<vendor>" on every inventory_items row and call rebuild_applications at the end of _run_sync.
Reconcile by making sure device rows expose a hostname that truncates to the same NetBIOS key and user rows expose UPN/email.
Env-seed — add settings fields + a _seed_<vendor>_config() awaited in lifespan.

For CSV sources, steps 1–3 collapse into: write one CsvProvider subclass and append it to _PROVIDERS. Full walkthroughs live in Extending recipes.