Recipes: extending the backend

Numbered, prescriptive recipes for extending the Lockstep VM Python backend. Every recipe lists the exact files to touch and shows the real code to write, grounded in the existing modules under webapp/backend/app.

Read these first for context:

Backend architecture — app structure, DI, lifespan, tenancy.
Backend style guide — naming, error handling, async rules.
Backend patterns — the integration lifecycle and adopt-on-read config.
Data model — collection shapes.

Non-negotiable conventions that every recipe depends on:

Every document carries a tenant_id string; every query filters on it.
current_user (from get_current_user) is a dict {user_id, email, tenant_id, role}. Read the tenant as current_user["tenant_id"].
Endpoints are thin HTTP adapters; reusable logic lives in services/.
Blocking I/O (HTTP clients, file parsing) runs in loop.run_in_executor(None, ...) so the event loop is never blocked.
Writes are idempotent: UpdateOne(..., upsert=True) keyed by a stable natural key, batched with bulk_write.

1. Add a new API integration source

Goal: add a source (call it Acme) that pulls assets/software from a REST API, follows the same lifecycle as Rapid7/SentinelOne/Intune (config → test-connection → sync → sync-status → stats), seeds from env on startup, and shows up as a page in the frontend.

Use the existing Rapid7 integration (api/v1/endpoints/rapid7.py) as your reference implementation — it is the canonical template.

Files to touch

#	File	Change
1	`webapp/backend/app/services/acme_client.py`	New client class (blocking HTTP).
2	`webapp/backend/app/api/v1/endpoints/acme.py`	New endpoint module (config/test/sync/status/stats).
3	`webapp/backend/app/api/v1/__init__.py`	Import + `include_router`.
4	`webapp/backend/app/core/config.py`	Add `ACME_*` settings.
5	`webapp/backend/app/main.py`	Add `_seed_acme_config()` and call it in `lifespan`.
6	`.env` / deployment env	Add `ACME_API_KEY`, `ACME_BASE_URL`.
7	`webapp/frontend/src/lib/api.ts` (or component) + `pages/acme.astro` + `components/AcmePage.tsx`	Frontend Sources page.
8	`webapp/frontend/src/components/Sidebar.tsx`	Sidebar entry under Sources.

Step 1 — client in `services/`

Clients have no FastAPI imports. They take plain constructor args and expose synchronous methods (they use httpx.Client, requests, or an SDK — all blocking). Model it on services/entra_client.py.

"""Acme REST client. Blocking httpx — call from endpoints via run_in_executor."""
from __future__ import annotations

import logging
from typing import Optional

import httpx

logger = logging.getLogger(__name__)


class AcmeAuthError(Exception):
    pass


class AcmeClient:
    def __init__(self, api_key: str, base_url: str, verbose: bool = False):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.verbose = verbose

    def _headers(self) -> dict:
        return {"Authorization": f"Bearer {self.api_key}", "Accept": "application/json"}

    def test_connection(self) -> bool:
        with httpx.Client(timeout=15) as client:
            resp = client.get(f"{self.base_url}/v1/ping", headers=self._headers())
        if resp.status_code == 401:
            raise AcmeAuthError("Acme returned 401 — check API key.")
        return resp.is_success

    def get_assets(self, fetch_all: bool = True) -> list[dict]:
        """Return a list of asset dicts. Page through the API here."""
        items: list[dict] = []
        url: Optional[str] = f"{self.base_url}/v1/assets?limit=500"
        with httpx.Client(timeout=30) as client:
            while url:
                resp = client.get(url, headers=self._headers())
                resp.raise_for_status()
                data = resp.json()
                items.extend(data.get("results", []))
                url = data.get("next") if fetch_all else None
        return items

Step 2 — endpoint module with the full lifecycle

Create api/v1/endpoints/acme.py. Copy the structure of rapid7.py exactly. The four load-bearing helpers are _resolve_config (adopt-on-read), _get_config_or_404, _safe_config, and _build_client.

"""Acme Integration Endpoints — config/test/sync/status/stats + background sync."""
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.acme_client import AcmeClient
from app.services.inventory_rollup import rebuild_applications

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


class AcmeConfigIn(BaseModel):
    api_key: str
    base_url: str
    enabled: bool = True


# ── Config resolution (adopt-on-read: env-seeded "default" applies to every tenant) ──
async def _resolve_config(tenant_id: str, db: AsyncIOMotorDatabase) -> Optional[dict]:
    config = await db.acme_config.find_one({"tenant_id": tenant_id})
    if config:
        return config
    if tenant_id != "default":
        seed = await db.acme_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.acme_config.update_one(
                {"tenant_id": tenant_id}, {"$setOnInsert": adopted}, upsert=True
            )
            return await db.acme_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="Acme not configured. POST /acme/config first.")
    return config


def _safe_config(doc: dict) -> dict:
    """Never leak the raw api_key back to the client."""
    return {
        "configured": True,
        "base_url": doc.get("base_url"),
        "has_key": bool(doc.get("api_key")),
        "sync_status": doc.get("sync_status"),
        "last_sync": doc.get("last_sync"),
        "sync_error": doc.get("sync_error"),
        "asset_count": doc.get("asset_count", 0),
        "app_count": doc.get("app_count", 0),
    }


def _build_client(config_doc: dict) -> AcmeClient:
    return AcmeClient(api_key=config_doc["api_key"], base_url=config_doc["base_url"])


# ── Background sync ───────────────────────────────────────────────────────────
async def _run_sync(tenant_id: str, db: AsyncIOMotorDatabase) -> None:
    try:
        await db.acme_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "running", "sync_started_at": datetime.utcnow()}},
        )
        config_doc = await db.acme_config.find_one({"tenant_id": tenant_id})
        client = _build_client(config_doc)
        now = datetime.utcnow()

        # Blocking HTTP → thread pool so we never block the event loop.
        loop = asyncio.get_running_loop()
        assets = await loop.run_in_executor(None, lambda: client.get_assets(fetch_all=True))

        live_ids: set[str] = set()
        asset_ops, inv_ops = [], []
        app_count = 0
        for a in assets:
            aid = str(a.get("id") or "")
            if not aid:
                continue
            live_ids.add(aid)
            asset_ops.append(UpdateOne(
                {"tenant_id": tenant_id, "asset_id": aid},
                {"$set": {
                    "tenant_id": tenant_id,
                    "asset_id": aid,
                    "hostname": a.get("hostname", ""),
                    "os_platform": a.get("os", ""),
                    "synced_at": now,
                }},
                upsert=True,
            ))
            # Installed software → inventory_items(source="acme") so it rolls up
            # into the Software inventory alongside every other source.
            for s in a.get("software", []):
                app_name = (s.get("name") or "").strip()
                if not app_name:
                    continue
                version = str(s.get("version") or "").strip()
                inv_ops.append(UpdateOne(
                    {"tenant_id": tenant_id, "source": "acme",
                     "asset_id": aid, "application_name": app_name, "version": version},
                    {"$set": {
                        "tenant_id": tenant_id, "source": "acme", "asset_id": aid,
                        "device_id": aid, "device_name": a.get("hostname", ""),
                        "application_name": app_name, "vendor": s.get("vendor", ""),
                        "version": version, "os_platform": a.get("os", ""),
                        "synced_at": now,
                    }},
                    upsert=True,
                ))
                app_count += 1

        if asset_ops:
            await db.acme_assets.bulk_write(asset_ops, ordered=False)
        if inv_ops:
            await db.inventory_items.bulk_write(inv_ops, ordered=False)

        # Remove stale rows no longer present in the source.
        if live_ids:
            live = [str(x) for x in live_ids]
            await db.acme_assets.delete_many({"tenant_id": tenant_id, "asset_id": {"$nin": live}})
            await db.inventory_items.delete_many(
                {"tenant_id": tenant_id, "source": "acme", "asset_id": {"$nin": live}}
            )

        # Rebuild the Software-inventory rollup so synced apps appear there.
        try:
            await rebuild_applications(tenant_id, db)
        except Exception as exc:
            logger.warning("Acme rollup failed (non-fatal): %s", exc)

        await db.acme_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {
                "sync_status": "success", "last_sync": now, "sync_error": None,
                "asset_count": len(assets), "app_count": app_count,
            }},
        )
    except Exception as exc:
        logger.error("Acme sync failed for tenant=%s: %s", tenant_id, exc)
        await db.acme_config.update_one(
            {"tenant_id": tenant_id},
            {"$set": {"sync_status": "error", "sync_error": str(exc)}},
        )


# ── Endpoints ─────────────────────────────────────────────────────────────────
@router.get("/config")
async def get_config(current_user: dict = Depends(get_current_user),
                     db: AsyncIOMotorDatabase = Depends(get_database)):
    doc = await _resolve_config(current_user["tenant_id"], db)
    return _safe_config(doc) if doc else {"configured": False}


@router.post("/config")
async def save_config(body: AcmeConfigIn,
                      current_user: dict = Depends(get_current_user),
                      db: AsyncIOMotorDatabase = Depends(get_database)):
    await db.acme_config.update_one(
        {"tenant_id": current_user["tenant_id"]},
        {"$set": {
            "tenant_id": current_user["tenant_id"],
            "api_key": body.api_key, "base_url": body.base_url,
            "enabled": body.enabled, "updated_at": datetime.utcnow(),
        }},
        upsert=True,
    )
    return {"configured": True}


@router.delete("/config")
async def delete_config(current_user: dict = Depends(get_current_user),
                        db: AsyncIOMotorDatabase = Depends(get_database)):
    tenant_id = current_user["tenant_id"]
    await db.acme_config.delete_one({"tenant_id": tenant_id})
    await db.acme_assets.delete_many({"tenant_id": tenant_id})
    await db.inventory_items.delete_many({"tenant_id": tenant_id, "source": "acme"})
    return {"deleted": True}


@router.post("/test-connection")
async def test_connection(current_user: dict = Depends(get_current_user),
                          db: AsyncIOMotorDatabase = Depends(get_database)):
    config_doc = await _get_config_or_404(current_user["tenant_id"], db)
    client = _build_client(config_doc)
    loop = asyncio.get_event_loop()
    try:
        ok = await loop.run_in_executor(None, client.test_connection)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc))
    if not ok:
        raise HTTPException(status_code=502, detail="Acme connection test failed — check API key/URL.")
    return {"status": "ok", "message": "Connection successful"}


@router.get("/sync-status")
async def sync_status(current_user: dict = Depends(get_current_user),
                      db: AsyncIOMotorDatabase = Depends(get_database)):
    doc = await _resolve_config(current_user["tenant_id"], db)
    if not doc:
        return {"sync_status": "not_configured"}
    status = doc.get("sync_status")
    started = doc.get("sync_started_at")
    # Stale-sync guard: a service restart can leave status stuck at "running".
    if status == "running" and started:
        try:
            age = (datetime.utcnow() - started).total_seconds()
        except TypeError:
            age = 0
        if age > 20 * 60:
            await db.acme_config.update_one(
                {"tenant_id": current_user["tenant_id"]},
                {"$set": {"sync_status": "error",
                          "sync_error": "Sync was interrupted (service restart). Click Sync to run again."}},
            )
            status = "error"
    return {
        "sync_status": status,
        "last_sync": doc.get("last_sync"),
        "sync_error": doc.get("sync_error"),
        "sync_started_at": doc.get("sync_started_at"),
    }


@router.post("/sync")
async def trigger_sync(background_tasks: BackgroundTasks,
                       current_user: dict = Depends(get_current_user),
                       db: AsyncIOMotorDatabase = Depends(get_database)):
    await _get_config_or_404(current_user["tenant_id"], db)
    background_tasks.add_task(_run_sync, current_user["tenant_id"], db)
    return {"status": "started"}


@router.get("/stats")
async def get_stats(current_user: dict = Depends(get_current_user),
                    db: AsyncIOMotorDatabase = Depends(get_database)):
    doc = await _resolve_config(current_user["tenant_id"], db)
    if not doc:
        return {"configured": False}
    return {
        "configured": True,
        "sync_status": doc.get("sync_status"),
        "last_sync": doc.get("last_sync"),
        "asset_count": doc.get("asset_count", 0),
        "app_count": doc.get("app_count", 0),
    }

Step 3 — register the router in `api/v1/init.py`

Add acme to the single import line, then add one include_router call. The URL prefix determines the frontend path (/api/v1/acme/...).

from app.api.v1.endpoints import (
    auth, inventory, ..., csv_sources, entra, acme,  # ← add acme
)

# ... existing includes ...
api_router.include_router(acme.router, prefix="/acme", tags=["acme"])

Step 4 — add settings in `core/config.py`

Add typed fields to the Settings class. All integration secrets default to "" so the app boots without them.

# webapp/backend/app/core/config.py  (inside class Settings)
# Acme
ACME_API_KEY: str = ""
ACME_BASE_URL: str = ""

Step 5 — seed from env in `main.py` lifespan

Add a _seed_acme_config() coroutine next to _seed_rapid7_config(), then call it in lifespan. The seed writes a tenant_id="default" config doc that _resolve_config later adopts for every tenant. It must be idempotent (skip if a default doc already exists).

async def _seed_acme_config() -> None:
    """Pre-populate Acme config for the default tenant from env vars if unset."""
    if not (settings.ACME_API_KEY and settings.ACME_BASE_URL):
        return
    try:
        db = get_database()
        existing = await db.acme_config.find_one({"tenant_id": "default"})
        if existing:
            logger.info("Acme config already exists for default tenant — skipping seed")
            return
        await db.acme_config.insert_one({
            "tenant_id": "default",
            "api_key": settings.ACME_API_KEY,
            "base_url": settings.ACME_BASE_URL,
            "enabled": True,
            "sync_status": None,
            "last_sync": None,
            "sync_error": None,
            "asset_count": 0,
            "app_count": 0,
            "updated_at": datetime.utcnow(),
        })
        logger.info("Acme config seeded from env vars (url=%s)", settings.ACME_BASE_URL)
    except Exception as exc:
        logger.error("Failed to seed Acme config: %s", exc)


@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()
    await _seed_acme_config()      # ← add this line
    yield
    await close_mongo_connection()

Step 6 — add env vars

# .env  (and your deployment secrets)
ACME_API_KEY=xxxxxxxxxxxxxxxx
ACME_BASE_URL=https://api.acme.example.com

Step 7 — frontend Sources page

Two files: a thin Astro route and a React component. Copy pages/rapid7.astro and components/Rapid7Page.tsx.

---
import Layout from '../layouts/Layout.astro';
import { AuthCheck } from '../components/AuthCheck';
import { AppLayout } from '../components/AppLayout';
import { AcmePage } from '../components/AcmePage';
---
<Layout title="Acme - Vulnerability Management">
  <AuthCheck client:load>
    <AppLayout client:load>
      <AcmePage client:load />
    </AppLayout>
  </AuthCheck>
</Layout>

The component talks to /acme/* with cookie-based fetch. Follow the React Query pattern in Rapid7Page.tsx (poll sync-status/stats on a refetchInterval while sync_status === 'running'). See Frontend pages & routes for the full page contract.

Step 8 — Sidebar entry

Add one NavItem to the Sources group in components/Sidebar.tsx (import a lucide-react icon at the top of the file):

// webapp/frontend/src/components/Sidebar.tsx  (Sources group)
{
  label: 'Sources',
  items: [
    { href: '/sentinelone', icon: ShieldIcon, label: 'SentinelOne' },
    { href: '/rapid7',      icon: Activity,   label: 'Rapid7' },
    { href: '/acme',        icon: Server,     label: 'Acme' },   // ← add
    // ...
  ],
},

Checklist

Software rows written to inventory_items with source="acme" and rebuild_applications called after sync.
_safe_config never returns the raw api_key.
Stale rows deleted with $nin against the live id set.
_seed_acme_config() added and called in lifespan.
Config resolution uses adopt-on-read so env credentials work for all tenants.

2. Add a new CSV provider

Goal: add a source that has no API and is fed by a CSV/Excel upload. The provider framework (services/csv_providers/) is fully generic — once you register a provider, its hub card, detail page, upload (“sync”), history, stats, and purge endpoints are automatic. You write exactly two things: a CsvProvider subclass and one line in the registry.

Files to touch

#	File	Change
1	`webapp/backend/app/services/csv_providers/providers.py`	New `CsvProvider` subclass.
2	`webapp/backend/app/services/csv_providers/__init__.py`	Register an instance in `_PROVIDERS`.

That’s it. api/v1/endpoints/csv_sources.py already exposes every route generically (GET /csv-sources/, GET /csv-sources/{key}, POST /csv-sources/{key}/upload, GET /csv-sources/{key}/uploads, DELETE /csv-sources/{key}/data, …), and the frontend CsvSourcesHub / CsvSourceDetail render every registered provider from its descriptor.

Pick a target

target decides which collection the parsed rows land in and how they are keyed. This is the single most important choice.

`target`	Collection written	Upsert key	Purpose
`software`	`inventory_items` (source=`key`)	`(tenant_id, source, device_id, application_name, version)`	Installed software; rolls up into the Software inventory.
`device`	`csv_devices`	`(tenant_id, source, device_key=hostname.lower())`	Device enrichment; links to an existing asset by hostname (replace-all per upload).
`user`	`csv_users`	`(tenant_id, source, email)`	User enrichment; links to an Entra user by email (replace-all per upload).
`records`	`csv_records`	inserted (replace-all per upload)	Non-software reference rows for tagging.

Your parse() must return records shaped for the chosen target (see the field names read by _process_upload in csv_sources.py):

software → device_id, device_name, application_name, vendor, version, os_platform, user_name.
device → device_name, attributes (dict).
user → email, display_name, site, attributes (dict), flags (dict).
records → any flat dict.

Step 1 — subclass `CsvProvider`

Set the class attributes (the descriptor that drives the UI card) and implement parse(rows, tenant_id) -> ParseResult. Use the loose column helpers self.pick(...) / self.pick_contains(...) so vendor exports work without pre-formatting. Model on the providers already in providers.py.

from .base import CsvProvider, ParseResult, normalize_platform


class JamfInventoryProvider(CsvProvider):
    """Jamf Pro 'Applications' export → software inventory."""

    key = "jamf_inventory"                 # URL slug + inventory_items `source`
    name = "Jamf Pro Inventory"
    description = "Jamf Pro installed-application export. One row per device + app. Feeds the Software inventory."
    category = "software"                  # software | identity | access (card grouping)
    target = "software"                    # software | device | user | records
    icon = "Package"                       # lucide icon name used by the card
    accepts = "Jamf 'Applications' CSV export"
    expected_columns = ["Computer Name", "Application Title", "Application Version", "Operating System"]
    # header_signals = ["Application Title"]  # set only if the export has a preamble block

    def parse(self, rows: list[dict], tenant_id: str) -> ParseResult:
        records, warnings = [], []
        for row in rows:
            app = self.pick(row, "application_title", "application", "app_name", "name")
            device = self.pick(row, "computer_name", "device_name", "hostname", "computer")
            if not app:
                continue
            records.append({
                "device_id": self.pick(row, "serial_number", "computer_name") or device,
                "device_name": device,
                "application_name": app,
                "vendor": self.pick(row, "vendor", "publisher", "developer"),
                "version": self.pick(row, "application_version", "version"),
                "os_platform": normalize_platform(self.pick(row, "operating_system", "os")),
                "user_name": self.pick(row, "username", "assigned_user"),
            })
        if not records:
            warnings.append("No rows had a recognizable application column.")
        return ParseResult(records=records, warnings=warnings, detected_shape="jamf")

Key base-class facts (services/csv_providers/base.py):

self.pick(row, *names) returns the first non-empty value among candidate column names, matched case/punctuation-insensitively.
self.pick_contains(row, *substrings) matches columns whose header contains a substring (for messy headers like "Supervisor's Name (First Last)").
load(content) auto-detects CSV vs XLSX by magic bytes and honours header_signals to skip a preamble — you do not override it.
ParseResult(records, warnings, detected_shape) — put user-visible parse notes in warnings; they surface on the upload detail page.

Step 2 — register in the registry

Append one instance to _PROVIDERS. The key becomes the registry map key, the URL slug, and (for software/records) the source field on written rows.

from .providers import (
    AutoElevateProvider,
    BadgeAccessProvider,
    HrRosterProvider,
    SoftwareInventoryProvider,
    JamfInventoryProvider,          # ← import
)

_PROVIDERS: List[CsvProvider] = [
    SoftwareInventoryProvider(),
    AutoElevateProvider(),
    BadgeAccessProvider(),
    HrRosterProvider(),
    JamfInventoryProvider(),        # ← register (card + endpoints now automatic)
]

No endpoint, no frontend, no seed, no config, no env changes. The card appears at /inventory (CSV hub) and the detail page at /csv-source/jamf_inventory immediately. See the CSV provider framework reference for the full descriptor/lifecycle contract.

3. Add a new REST endpoint to an existing router

Goal: add a route to an existing endpoint module (e.g. a GET /rapid7/summary). Every route follows the same DI shape: resolve current_user and db, filter by tenant_id, project out _id, return plain dict/JSON.

File to touch

webapp/backend/app/api/v1/endpoints/<module>.py — add a @router.<method> function. The prefix is already set in api/v1/__init__.py; do not touch it unless you are adding a whole new module (see Recipe 1, Step 3).

Pattern

# webapp/backend/app/api/v1/endpoints/rapid7.py  (append a route)
from typing import Optional
from fastapi import Query

@router.get("/summary")
async def get_summary(
    min_risk: int = Query(0, ge=0),                 # typed query param + validation
    search: Optional[str] = None,
    current_user: dict = Depends(get_current_user), # auth + tenant
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    tenant_id = current_user["tenant_id"]
    match: dict = {"tenant_id": tenant_id, "risk_score": {"$gte": min_risk}}
    if search:
        match["hostname"] = {"$regex": search, "$options": "i"}

    # Always project out _id (ObjectId is not JSON-serializable).
    total = await db.rapid7_assets.count_documents(match)
    top = await (
        db.rapid7_assets.find(match, {"_id": 0})
        .sort("risk_score", -1).limit(10).to_list(length=10)
    )
    # Convert datetimes to ISO strings before returning (Mongo returns datetime).
    for d in top:
        for k, v in list(d.items()):
            if hasattr(v, "isoformat"):
                d[k] = v.isoformat()
    return {"total": total, "top_assets": top}

Rules (from the style guide):

Rule	Why
First filter key is always `tenant_id`	Multi-tenant isolation; never query without it.
`current_user: dict = Depends(get_current_user)`	Auth. For admin-only routes use `Depends(get_current_admin_user)`.
`db: AsyncIOMotorDatabase = Depends(get_database)`	Async Mongo handle.
Project `{"_id": 0}`	`ObjectId` is not JSON-serializable.
POST bodies use a Pydantic `BaseModel`	Validation + typed contract.
Raise `HTTPException(status_code=..., detail=...)`	Consistent error surface.
`await` every Mongo call	Motor is async; a missing `await` returns a coroutine.
Blocking work → `loop.run_in_executor(None, fn)`	Never block the event loop.
Long work (>1s) → `background_tasks.add_task(...)`	Return `{"status": "started"}` immediately.

For a request body:

from pydantic import BaseModel

class TagBody(BaseModel):
    asset_id: str
    tags: list[str]

@router.post("/tag")
async def tag_asset(
    body: TagBody,
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    result = await db.rapid7_assets.update_one(
        {"tenant_id": current_user["tenant_id"], "asset_id": body.asset_id},
        {"$set": {"tags": body.tags}},
    )
    if result.matched_count == 0:
        raise HTTPException(status_code=404, detail="Asset not found")
    return {"updated": True}

4. Add a query-builder field

Goal: expose a new filterable field in the Assets or Software query builder. Fields are declared in allowlists in services/query_builder.py; the builder can only reference paths in the allowlist, so adding a field is a one-line, safe change.

File to touch

webapp/backend/app/services/query_builder.py — add an entry to APPLICATION_FIELDS (Software) or DEVICE_FIELDS (Assets).

How it works

Each entry maps a logical field name (what the UI shows) to a FieldSpec(db_field=..., type=...). type is one of "string" | "number" | "boolean" | "date" and controls operator coercion (e.g. equals on a string becomes a case-insensitive anchored regex; on a number it’s an exact match).

# Software fields (queried against the `applications` rollup):
APPLICATION_FIELDS: Dict[str, FieldSpec] = {
    "name":         FieldSpec(db_field="name",         type="string"),
    "vendor":       FieldSpec(db_field="vendor",       type="string"),
    "device_count": FieldSpec(db_field="device_count", type="number"),
    # ── add your field ──
    "license_type": FieldSpec(db_field="license_type", type="string"),
}

# Assets fields (queried against the /devices/search aggregation):
DEVICE_FIELDS: Dict[str, FieldSpec] = {
    "device_name": FieldSpec(db_field="device_name", type="string"),
    "os_platform": FieldSpec(db_field="os_platform", type="string"),
    # Source-joined fields use a dotted prefix that matches the $lookup alias
    # in /devices/search (intune.*, s1.*, r7.*):
    "r7.risk_score": FieldSpec(db_field="r7.risk_score", type="number"),
    # ── add your field ──
    "acme.serial":   FieldSpec(db_field="acme.serial",   type="string"),
}

Rules

The db_field must actually exist on the document being searched. For APPLICATION_FIELDS that is the applications rollup doc (see rebuild_applications in services/inventory_rollup.py) — if you add a field, make sure the rollup writes it. For DEVICE_FIELDS, shared fields come from the inventory_items aggregation and prefixed fields (intune., s1., r7.) come from $lookup stages in /devices/search; a new prefix (e.g. acme.) requires a matching $lookup + alias in that endpoint’s pipeline.
Unknown fields are silently skipped by _rule_to_mongo (safer than 400-ing the whole search), so a typo produces no error — verify the field returns results.
The frontend field catalog (components/dashboard/fieldCatalogs.ts and the query-builder UI) should list the same logical name so users can pick it.
APPLICATION_FIELDS is extended in two places in the file (a base dict plus an .update({...}) call) — append to either.

5. Add a new MongoDB collection

Goal: introduce a new collection. There is no ORM and no migration system — collections are created implicitly on first write. What matters is following the document conventions so the collection is tenant-safe and idempotent.

There are no files that “declare” collections

You access a collection simply as an attribute of the db handle: db.acme_assets. It springs into existence on the first insert/upsert. So “adding a collection” means: write documents that follow the conventions, and (optionally) declare indexes.

Conventions (mandatory)

Convention	Rule	Example
Tenant scoping	Every document has a `tenant_id` string; every query and every upsert filter includes it as the first key.	`{"tenant_id": tenant_id, ...}`
Stable upsert key	Choose a natural key that uniquely identifies a row within a tenant+source and use it in the `UpdateOne` filter with `upsert=True`. Never rely on `_id`.	`{"tenant_id": t, "source": "acme", "asset_id": aid}`
`source` tag	For any collection fed by multiple integrations (like `inventory_items`), tag rows with `source=<key>` so they can be rebuilt/purged per source.	`"source": "acme"`
`synced_at` / `updated_at`	Stamp `datetime.utcnow()` on every write so stale rows can be detected and the UI can show freshness.	`"synced_at": now`
Stale cleanup	After a full sync, delete rows not in the live set: `delete_many({"tenant_id": t, "source": k, "<key>": {"$nin": live_ids}})`.	see Recipe 1 sync
No `_id` leakage	Project `{"_id": 0}` on reads; `ObjectId` is not JSON-serializable.	`.find(q, {"_id": 0})`
Config collections	A `<source>_config` collection holds one doc per tenant with `tenant_id`, credentials, `sync_status`, `last_sync`, `sync_error`, and counts. Seeded for `tenant_id="default"` and adopted on read.	`db.acme_config`

The canonical write

from datetime import datetime
from pymongo import UpdateOne

now = datetime.utcnow()
ops = [
    UpdateOne(
        {"tenant_id": tenant_id, "source": "acme", "asset_id": aid},   # upsert key
        {"$set": {
            "tenant_id": tenant_id,          # always store it, not just filter on it
            "source": "acme",
            "asset_id": aid,
            "hostname": hostname,
            "synced_at": now,
        }},
        upsert=True,
    )
    for aid, hostname in rows
]
# Batch in chunks — Mongo caps bulk_write size; the codebase uses 1000.
for i in range(0, len(ops), 1000):
    await db.acme_assets.bulk_write(ops[i:i + 1000], ordered=False)

Use ordered=False so one bad op doesn’t abort the batch. For “the fresh export is the full set” semantics (device/user enrichment), the pattern is delete_many({tenant_id, source}) then insert — see _process_upload in csv_sources.py.

Indexes

The codebase currently creates no indexes programmatically (mongodb-init/ is empty and there are no create_index calls); it relies on MongoDB defaults and always-tenant-filtered queries. If a new collection is large and hot, add a compound index whose first field is tenant_id, matching your most common query, in a startup task. Put it alongside connect_to_mongo so it runs once at boot:

# e.g. inside connect_to_mongo() in webapp/backend/app/core/database.py
await db.db.acme_assets.create_index(
    [("tenant_id", 1), ("asset_id", 1)], unique=True, name="tenant_asset"
)
await db.db.acme_assets.create_index([("tenant_id", 1), ("risk_score", -1)])

The unique compound index should match your upsert key so duplicates are impossible. Order matters: put the equality field (tenant_id) first, then the sort/range field.

Data model — existing collection shapes (inventory_items, applications, *_config, csv_*).
Backend architecture — the db handle and Motor access.
Backend patterns — adopt-on-read config and the sync lifecycle.

Recipes: extending the backend

1. Add a new API integration source

Files to touch

Step 1 — client in services/

Step 2 — endpoint module with the full lifecycle

Step 3 — register the router in api/v1/__init__.py

Step 4 — add settings in core/config.py

Step 5 — seed from env in main.py lifespan

Step 6 — add env vars

Step 7 — frontend Sources page

Step 8 — Sidebar entry

Checklist

2. Add a new CSV provider

Files to touch

Pick a target

Step 1 — subclass CsvProvider

Step 2 — register in the registry

3. Add a new REST endpoint to an existing router

File to touch

Pattern

4. Add a query-builder field

File to touch

How it works

Rules

5. Add a new MongoDB collection

There are no files that “declare” collections

Conventions (mandatory)

The canonical write

Indexes

Related reading

Step 1 — client in `services/`

Step 3 — register the router in `api/v1/init.py`

Step 4 — add settings in `core/config.py`

Step 5 — seed from env in `main.py` lifespan

Step 1 — subclass `CsvProvider`