Backend style guide

This is the coding standard for the Lockstep VM backend (webapp/backend/app). It is written for AI coding agents adding or editing endpoints and services. Every rule here is extracted from the real integration modules — intune.py, rapid7.py, connectwise_automate.py, entra.py, csv_sources.py, devices.py, inventory.py — and the shared services they call. Follow these patterns exactly. Do not invent new shapes; copy an existing sibling module and adapt it.

The stack is FastAPI + Motor (async MongoDB) + Pydantic. See also Data model for collection shapes and Backend reference for the module map.

Module skeleton

Every integration endpoint module follows the same section order. Match it, including the # ── … ── comment banners, so files are diffable against each other.

"""
<Provider> integration endpoints.

One-paragraph description of what the module pulls and where it writes.
"""
from __future__ import annotations

import asyncio
import logging
from datetime import datetime
from typing import Optional

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from motor.motor_asyncio import AsyncIOMotorDatabase
from pydantic import BaseModel
from pymongo import UpdateOne

from app.core.database import get_database
from app.core.security import get_current_user
from app.services.<provider>_client import ProviderClient, ProviderAuthError, ProviderError
from app.services.inventory_rollup import rebuild_applications

logger = logging.getLogger(__name__)
router = APIRouter()

SOURCE = "connectwise_automate"   # module-level constant when the collection tag is reused

# ── Models ───────────────────────────────────────────────────────────────────
# ── Helpers ──────────────────────────────────────────────────────────────────
# ── Background sync ──────────────────────────────────────────────────────────
# ── Routes ───────────────────────────────────────────────────────────────────

Rules:

Always from __future__ import annotations as the first import so list[str], dict[str, dict] and Optional annotations work uniformly.
Always logger = logging.getLogger(__name__) and router = APIRouter() at module top. Never configure logging in the module.
Import get_database and get_current_user from app.core.*; import the vendor client and its exceptions from app.services.<provider>_client.
Do the provider SDK import lazily inside _build_client when the client lives outside app.services (e.g. Rapid7 imports from vulnmgmt.integrations.rapid7.client import Rapid7Client inside the helper). App-service clients are imported at module top.

Async / await everywhere

Every route handler and every helper that touches Mongo is async def. Every Motor call is awaited.

@router.get("/stats")
async def get_stats(
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    tenant_id = current_user["tenant_id"]
    device_count = await db.intune_devices.count_documents({"tenant_id": tenant_id})

Never call blocking I/O directly on the event loop. Vendor SDK HTTP calls are synchronous — push them into the default thread-pool executor:

loop = asyncio.get_event_loop()
computers = await loop.run_in_executor(None, client.get_all_computers)

When you need to pass arguments, wrap in a closure (bind loop-variant values as default args to avoid late-binding bugs):

def _fetch_computers():
    if scope_ids:
        out = []
        for cid in scope_ids:
            out.extend(client.get_computers_for_client(cid))
        return out
    return client.get_all_computers(page_size=500)

computers = await loop.run_in_executor(None, _fetch_computers)

def _fetch_software_batch(batch_ids=batch):   # bind `batch` as a default arg
    ...
fetched = await loop.run_in_executor(None, _fetch_software_batch)

Run independent counts concurrently with asyncio.gather rather than awaiting serially:

intune_count, sentinelone_count, rapid7_count = await asyncio.gather(
    db.intune_devices.count_documents({"tenant_id": tid}),
    db.sentinelone_devices.count_documents({"tenant_id": tid}),
    db.rapid7_assets.count_documents({"tenant_id": tid}),
)

Pydantic request models

Define a BaseModel per POST body, named <Provider>Config or <Thing>ConfigIn. Keep secrets Optional so the UI can save non-secret edits without re-sending them. Default non-required scalars inline.

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

Bind the model as the first parameter (before the Depends args): async def configure(config: IntuneConfig, current_user: dict = Depends(...), ...). For the shared structured-search endpoints, reuse SearchRequest from app.services.query_builder — do not roll your own.

Cross-field validation that Pydantic can’t express goes at the top of the handler as an HTTPException(422):

if config.auth_method == "secret" and not config.client_secret:
    raise HTTPException(status_code=422, detail="client_secret required for auth_method='secret'")

Route decorators and dependencies

Sub-path only in the decorator (@router.post("/config")); the prefix is set where the router is mounted.
Every handler takes current_user: dict = Depends(get_current_user) and db: AsyncIOMotorDatabase = Depends(get_database) as the last two parameters. There are no exceptions — an unauthenticated route does not exist in this codebase.
For admin-only routes depend on get_current_admin_user instead (from app.core.security).
Query params are plain typed function args with defaults; use fastapi.Query when you need bounds (Query(0, ge=0), Query(100, ge=1, le=1000)).
Set status_code=201 on create/upload routes: @router.post("/{key}/upload", status_code=201).
Add a one-line docstring describing the route.

@router.get("/devices")
async def list_devices(
    skip: int = 0,
    limit: int = 100,
    os_filter: Optional[str] = None,
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    """List synced Intune managed devices."""
    query: dict = {"tenant_id": current_user["tenant_id"]}
    ...

Tenant scoping — non-negotiable

current_user["tenant_id"] MUST appear in every Mongo filter — find, find_one, count_documents, delete_*, update_*, and the $match stage of every aggregation. This is the security boundary; a query without it leaks another tenant’s data.

query: dict = {"tenant_id": current_user["tenant_id"]}
if search:
    query["$or"] = [
        {"hostname": {"$regex": search, "$options": "i"}},
        {"ip_address": {"$regex": search, "$options": "i"}},
    ]
total = await db.rapid7_assets.count_documents(query)

Aggregations start with a tenant $match:

os_pipeline = [
    {"$match": {"tenant_id": tenant_id}},
    {"$group": {"_id": "$os", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}},
]

Inside $lookup sub-pipelines the join must also re-assert the tenant, because a lookup crosses collections:

{"$lookup": {
    "from": "intune_devices",
    "let": {"n": "$_name_lc"},
    "pipeline": [
        {"$match": {"tenant_id": tenant_id}},                       # tenant first
        {"$match": {"$expr": {"$eq": [{"$toLower": "$device_name"}, "$$n"]}}},
        {"$project": {"_id": 0}},
        {"$limit": 1},
    ],
    "as": "intune",
}}

Convention: bind tenant_id = current_user["tenant_id"] to a local at the top of the handler when you reference it more than once; inline current_user["tenant_id"] when it is used once.

The four config helpers

Every integration module defines the same four helpers over its <provider>_config collection. Copy them verbatim and rename the collection. They are the contract the routes are built on.

Helper	Signature	Job
`_build_client`	`(config_doc: dict) -> ProviderClient`	Instantiate the vendor SDK client from a stored config doc. Pure, sync.
`_resolve_config`	`async (tenant_id, db) -> Optional[dict]`	Return this tenant’s config, adopting the env-seeded `"default"` config on first access.
`_get_config_or_404`	`async (tenant_id, db) -> dict`	`_resolve_config` or raise `HTTPException(404)`.
`_safe_config`	`(config_doc: dict) -> dict`	Config with all secrets redacted for API responses.

_build_client reads with .get() for optional fields, [...] for required ones:

def _build_client(config_doc: dict) -> IntuneClient:
    """Instantiate IntuneClient from a stored config document."""
    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"),
    )

_resolve_config implements the default-config adoption pattern — env-seeded credentials under tenant_id="default" apply to every tenant until that tenant edits or deletes its own copy. Copy this exactly:

async def _resolve_config(tenant_id: str, db: AsyncIOMotorDatabase) -> Optional[dict]:
    """Return this tenant's config, adopting the env-seeded 'default' config on
    first access (per-tenant edits/deletes win afterwards)."""
    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

_get_config_or_404 gives a directive error detail (tell the caller how to fix it):

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

Never leak secrets — the `_safe_config` rule

No route may ever return a config document directly. Secrets (client_secret, cert_private_key, api_key, password) live encrypted in Mongo and MUST NOT appear in any HTTP response. Every GET on a config collection returns _safe_config(doc), which allowlists non-secret fields and replaces each secret with a boolean has_* presence flag.

def _safe_config(config_doc: dict) -> dict:
    """Return config doc with secrets redacted for API responses."""
    return {
        "azure_tenant_id": config_doc.get("azure_tenant_id"),
        "azure_client_id": config_doc.get("azure_client_id"),
        "auth_method": config_doc.get("auth_method", "secret"),
        "cert_thumbprint": config_doc.get("cert_thumbprint"),
        "enabled": config_doc.get("enabled", True),
        "has_secret": bool(config_doc.get("client_secret")),   # ← presence flag, never the value
        "has_cert": bool(config_doc.get("cert_private_key")),  # ← presence flag, never the value
        "last_sync": config_doc.get("last_sync"),
        "sync_status": config_doc.get("sync_status"),
        "sync_error": config_doc.get("sync_error"),
        "device_count": config_doc.get("device_count", 0),
        "updated_at": config_doc.get("updated_at"),
    }

@router.get("/config")
async def get_intune_config(current_user: dict = Depends(get_current_user), db=Depends(get_database)):
    config = await _resolve_config(current_user["tenant_id"], db)
    if not config:
        return {"configured": False}
    return {"configured": True, **_safe_config(config)}   # spread, never `return config`

_safe_config is allowlist, not denylist — it names the fields to include. If you add a new secret field to a config model, it is redacted by default (you must not add it to _safe_config). Never build a “redact these keys” denylist.

On save, only write a secret when a new value is supplied, so a partial update from the UI doesn’t wipe the stored credential:

doc = {"tenant_id": tenant_id, "username": config.username, "updated_at": datetime.utcnow(), ...}
if config.password:                 # only overwrite when a new secret is supplied
    doc["password"] = config.password
await db.connectwise_automate_config.update_one({"tenant_id": tenant_id}, {"$set": doc}, upsert=True)

Background sync tasks

Long-running syncs run via BackgroundTasks, never inline in the request. The sync coroutine is named _run_sync(tenant_id, db) and lives in the # ── Background sync ── section.

@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":     # idempotency guard
        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"}

_run_sync follows a fixed lifecycle. Set status running first, wrap everything in try/except, and record success or error with timestamps on the config doc:

async def _run_sync(tenant_id: str, db: AsyncIOMotorDatabase) -> None:
    now = datetime.utcnow()
    try:
        await db.provider_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "running", "sync_started_at": now, "sync_error": None}},
        )
        config_doc = await db.provider_config.find_one({"tenant_id": tenant_id})
        client = _build_client(config_doc)
        loop = asyncio.get_event_loop()

        data = await loop.run_in_executor(None, client.get_everything)   # blocking HTTP off-loop
        # ... bulk upsert, stale cleanup, rebuild_applications ...

        await db.provider_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "success", "last_sync": now, "sync_error": None,
                      "device_count": ..., "app_count": ...}},
        )
        logger.info("Provider sync complete tenant=%s: %d devices", tenant_id, ...)
    except ProviderAuthError as exc:                       # specific first
        logger.error("Provider sync auth failed tenant=%s: %s", tenant_id, exc)
        await db.provider_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "error", "sync_error": f"Authentication failed: {exc}"}},
        )
    except Exception as exc:                                 # catch-all last
        logger.exception("Provider sync failed tenant=%s", tenant_id)
        await db.provider_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "error", "sync_error": str(exc)}},
        )

Optional but recommended patterns seen in the codebase:

Progress reporting — update a sync_progress sub-doc per batch so the UI can poll (rapid7.py).
Stale-sync guard — in sync-status, if status has been running with no update for >20 min, flip it to error so a backend restart doesn’t wedge the UI (rapid7.py).
Circuit breaker — when an optional sub-resource (per-asset /software) keeps 404-ing, disable it after SW_FAIL_LIMIT = 15 consecutive failures and log one warning instead of hammering every asset.

Motor / MongoDB patterns

find / find_one with projections

Always project. Exclude _id with {"_id": 0} when you’re returning the doc as JSON (ObjectId is not JSON-serializable). Use a positive projection to fetch only the fields you need:

cursor = db.rapid7_assets.find(query, {"_id": 0}).skip(skip).limit(limit).sort("risk_score", -1)
assets = await cursor.to_list(length=limit)

async for doc in db.intune_devices.find(
    {"tenant_id": tenant_id},
    {"_id": 0, "device_name": 1, "intune_device_id": 1, "user_email": 1},   # positive projection
):
    ...

When you keep _id (list endpoints that expose an id), stringify it into an id field and pop the raw ObjectId:

for d in devices:
    d["id"] = str(d.pop("_id"))

Iterate large result sets with async for; use .to_list(length=...) when you need a bounded list. Always pass an explicit length.

Pagination and search

Clamp pagination defensively — never trust the client’s skip/limit:

.skip(max(0, skip)).limit(max(1, min(limit, 500)))

Free-text search is a case-insensitive $regex across a small set of fields, $or-combined:

if search:
    query["$or"] = [
        {"device_name": {"$regex": search, "$options": "i"}},
        {"user_email":  {"$regex": search, "$options": "i"}},
    ]

re.escape() any user string interpolated into an anchored regex (exact-match lookups) so metacharacters can’t break the pattern:

{"upn": {"$regex": f"^{_re.escape(upn)}$", "$options": "i"}}

count_documents

Use count_documents(query) for the total in a paginated response — the same query as the find. For aggregated/grouped counts, run a parallel $count pipeline built from the same stages:

total = await db.intune_devices.count_documents(query)
cursor = db.intune_devices.find(query).skip(skip).limit(limit).sort("device_name", 1)

count_pipeline = base_stages + [{"$count": "total"}]
count_res = await db.inventory_items.aggregate(count_pipeline).to_list(length=1)
total = count_res[0]["total"] if count_res else 0

aggregate pipelines

Build pipelines as list[dict]. First stage is always the tenant $match. Read results with .to_list(length=...) or async for. Pass allowDiskUse=True for large group-bys.

sev = await db.rapid7_assets.aggregate([
    {"$match": {"tenant_id": tenant_id}},
    {"$group": {"_id": None,
                "critical": {"$sum": "$critical_vulnerabilities"},
                "severe":   {"$sum": "$severe_vulnerabilities"}}},
]).to_list(length=1)
critical_total = sev[0]["critical"] if sev else 0     # guard empty result

Reuse one base_stages list for both the count and the data pipeline so they can never diverge (devices.py search_devices):

count_pipeline = base_stages + [{"$count": "total"}]
data_pipeline  = base_stages + [{"$sort": {sort_field: sort_dir}}, {"$skip": skip},
                                {"$limit": limit}, {"$project": {"_id": 0, "_name_lc": 0}}]

Normalize null/missing group keys in Python when projecting out (r["_id"] or "Unknown"), and use {"$ifNull": ["$source", "intune"]} inside the pipeline for legacy rows that predate a field.

bulk_write with UpdateOne upserts

Sync writes are always UpdateOne(..., upsert=True) batched into a single bulk_write(ops, ordered=False). Never loop update_one per record. The filter is the natural key {tenant_id, <source key(s)>}; the $set payload re-includes tenant_id and stamps synced_at.

from pymongo import UpdateOne

dev_ops = []
for device in devices:
    did = device.get("intune_device_id")
    dev_ops.append(UpdateOne(
        {"tenant_id": tenant_id, "intune_device_id": did},       # natural-key filter (tenant-scoped)
        {"$set": {**device, "tenant_id": tenant_id, "synced_at": now}},
        upsert=True,
    ))
if dev_ops:
    await db.intune_devices.bulk_write(dev_ops, ordered=False)

Chunk large op lists to 1000 per bulk_write to avoid oversized batches:

for i in range(0, len(inv_ops), 1000):
    await db.inventory_items.bulk_write(inv_ops[i:i + 1000], ordered=False)

Always guard if ops: before calling — bulk_write([]) raises.

Stale-record cleanup

After upserting, delete rows no longer present in the source using $nin on the set of live ids you collected during the loop (tenant + source scoped):

live_device_ids: set[str] = set()
# ... populate during the upsert loop ...
if live_device_ids:
    stale = await db.intune_devices.delete_many(
        {"tenant_id": tenant_id, "intune_device_id": {"$nin": list(live_device_ids)}}
    )
    if stale.deleted_count:
        logger.info("Removed %d stale Intune devices for tenant=%s", stale.deleted_count, tenant_id)

For records/user/device-target CSV imports where a fresh export is the full set, replace instead: delete_many({tenant_id, source}) then bulk-insert.

Writing into the shared `inventory_items` collection

Any source that reports installed software writes one inventory_items row per (device, app, version) with a source tag, then calls rebuild_applications so it rolls up into the Software inventory. The upsert key and field set are standardized — match them so cross-source dedupe works:

inv_ops.append(UpdateOne(
    {"tenant_id": tenant_id, "source": SOURCE,
     "device_id": cid, "application_name": app_name, "version": version},
    {"$set": {
        "tenant_id": tenant_id,
        "source": SOURCE,
        "device_id": cid,
        "device_name": meta.get("device_name", ""),
        "application_name": app_name,
        "vendor": (s.get("Publisher") or "").strip(),
        "version": version,
        "os_platform": meta.get("os_platform", ""),
        "synced_at": now,
    }},
    upsert=True,
))

Then, non-fatally:

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

DELETE /config and DELETE /{key}/data routes must also purge this source’s inventory_items rows and re-run the rollup. See Data model for the inventory_items and applications shapes.

Error handling

Raise fastapi.HTTPException(status_code=..., detail="..."); never return an ad-hoc error dict from a route.
Catch the vendor’s specific exceptions first, then a catch-all Exception last.
Re-raise HTTPException before a generic except so you don’t wrap a 404 into a 502:

try:
    ok = await loop.run_in_executor(None, client.test_connection)
    if not ok:
        raise HTTPException(status_code=502, detail="Rapid7 connection test failed — check API key and region")
    return {"status": "ok", "message": "Connection successful"}
except HTTPException:
    raise                                             # let intended HTTP errors through
except Exception as exc:
    raise HTTPException(status_code=502, detail=str(exc))

Status-code conventions:

Code	When
`400`	Malformed input the model can’t reject (bad ObjectId, unparseable id).
`401`	Vendor auth failed (`ProviderAuthError`).
`404`	Config missing (`_get_config_or_404`) or record not found.
`422`	Cross-field validation the Pydantic model can’t express.
`502`	Upstream vendor API error (`ProviderError` / unexpected error on a live call).

try:
    client = _build_client(cfg)
    return await loop.run_in_executor(None, client.test_connection)
except EntraAuthError as exc:
    raise HTTPException(status_code=401, detail=str(exc))
except EntraAPIError as exc:
    raise HTTPException(status_code=502, detail=str(exc))

Validate an ObjectId before using it:

try:
    oid = ObjectId(upload_id)
except Exception:
    raise HTTPException(status_code=400, detail="Bad upload id")

In background tasks there is no HTTP response — never raise out of _run_sync. Swallow the exception, log it, and persist sync_status="error" with sync_error on the config doc (see the sync lifecycle above).

Logging

Use the module logger. Use %-style lazy args, not f-strings, in log calls. Level guide:

Call	Use for
`logger.info(...)`	Successful milestones — fetch counts, batch completion, sync complete, stale removals.
`logger.warning(...)`	Recoverable / non-fatal degradation — rollup failed, sub-resource disabled by circuit breaker.
`logger.error(...)`	A handled failure you’re reporting to the caller (e.g. vendor sync failed, status set to error).
`logger.exception(...)`	An unexpected exception where you want the traceback — the catch-all `except Exception` in a sync.
`logger.debug(...)`	Per-item best-effort failures inside a loop (e.g. one group’s members failed).

logger.info("Entra: fetched %d users, %d groups for tenant=%s", len(users), len(groups), tenant_id)
logger.warning("CW Automate applications rollup failed (non-fatal): %s", rollup_exc)
logger.error("Entra sync auth failed tenant=%s: %s", tenant_id, exc)
logger.exception("Unexpected Intune sync error for tenant=%s", tenant_id)

Include tenant_id in sync/log lines so multi-tenant logs are greppable. Do not print() for real logging — the few print(..., flush=True) lines in rapid7.py are temporary sync-debug scaffolding, not the standard; do not add more.

Naming conventions

Convention	Example
Private module helpers are prefixed `_`	`_build_client`, `_resolve_config`, `_safe_config`, `_run_sync`, `_get_config_or_404`, `_normalize_platform`, `_short_hostname`
Config Pydantic model	`IntuneConfig`, `Rapid7ConfigIn`, `CWAutomateConfig`
Vendor client factory	`_build_client(config_doc) -> ProviderClient`
Vendor exceptions	`ProviderAuthError`, `ProviderError` / `ProviderAPIError`
Background sync coroutine	`_run_sync(tenant_id, db)`
Reused collection-tag constant	`SOURCE = "connectwise_automate"`
Collections	snake_case: `intune_devices`, `rapid7_assets`, `inventory_items`, `<provider>_config`
Timestamp fields	`synced_at`, `last_sync`, `sync_started_at`, `created_at`, `updated_at`
Status fields	`sync_status` ∈ `{"running","success","error"}`, `sync_error`, `sync_progress`

Nested helper closures that run in the executor are named _fetch_* (_fetch_all, _fetch_computers, _fetch_software_batch, _fetch_memberships).

Date & time handling

Store timestamps as datetime.utcnow() (naive UTC) — this is the codebase-wide convention. Compute now = datetime.utcnow() once at the top of a sync and reuse it for every synced_at in that run so a batch shares a single timestamp.
Use datetime.now().strftime("%Y%m%d_%H%M%S") only for filesystem-safe filename stamps (upload paths), not for stored data.
When returning a doc that may contain datetime values as JSON, convert them with isoformat():

for k, v in list(doc.items()):
    if hasattr(v, "isoformat"):
        doc[k] = v.isoformat()

Age math for the stale-sync guard: (datetime.utcnow() - started).total_seconds(), wrapped in try/except TypeError in case started is missing.

Response shapes

Return plain dicts (FastAPI serializes them). Conventions:

List endpoints return {"total": <count>, "skip": skip, "limit": limit, "<plural>": rows}.
Config GET returns {"configured": bool, **_safe_config(doc)} (or {"configured": False} when absent).
Save/delete return a small ack: {"configured": True, "message": "..."}, {"deleted": True}.
A response_model (Pydantic) is optional and used sparingly (inventory.py list_inventory_items); most routes return raw dicts.

Do / Don’t

Put {"tenant_id": current_user["tenant_id"]} in every single query, aggregation $match, and $lookup sub-pipeline.
Return _safe_config(doc) (or a hand-built allowlist) from every config-reading route.
await every Motor call; push blocking vendor SDK calls into run_in_executor.
Batch writes with bulk_write([UpdateOne(...upsert=True)], ordered=False), chunked to 1000, guarded by if ops:.
Project {"_id": 0} (or stringify _id → id) before returning docs as JSON.
Clamp skip/limit with max(0, skip) / max(1, min(limit, 500)).
Raise HTTPException with a specific status; re-raise HTTPException before any catch-all except.
Log with %-args and include tenant_id; use logger.exception in the sync catch-all.
Copy an existing sibling module (rapid7.py, intune.py) as the template for a new integration.
Call rebuild_applications(tenant_id, db) (non-fatally) after any write to inventory_items, and purge it on delete.

Don’t

Don’t run a query without tenant_id — it is a cross-tenant data leak.
Don’t return config (or any doc containing client_secret / cert_private_key / api_key / password) from a route.
Don’t build a redaction denylist — allowlist the safe fields in _safe_config.
Don’t overwrite a stored secret when the incoming value is empty (if config.password: guard).
Don’t call blocking HTTP on the event loop, and don’t time.sleep/requests directly in a handler.
Don’t loop update_one per record where a batched bulk_write works.
Don’t raise out of _run_sync / background tasks — persist sync_status="error" instead.
Don’t run the sync inline in the request handler — dispatch via BackgroundTasks.
Don’t use f-strings inside logger.* calls; don’t add print() logging.
Don’t return a bare datetime in JSON without isoformat(), and don’t return a raw ObjectId.
Don’t invent new field/collection/status names — reuse the shared vocabulary (synced_at, sync_status, source, inventory_items).

Reference

Backend reference — module map and mounting.
Data model — collection schemas (*_config, *_devices, inventory_items, applications).
Engine: integrations library — the vulnmgmt.integrations.* vendor SDK clients that _build_client wraps.
Integration guides: Intune · Rapid7 · ConnectWise Automate · Entra · CSV providers.