CSV provider framework

Not every tool an MSP runs has an API. Privilege-management consoles, physical badge systems, HR platforms and one-off vendor exports often only give you a CSV or Excel download. The CSV provider framework brings those sources into the same reconciled inventory as the API integrations — without special cases in the endpoints, the upload jobs, or the UI.

The idea: parser as “client”, upload as “sync”

The framework deliberately mirrors the API-integration lifecycle (client → config → sync → data → rollup). The mapping is:

API integration	CSV provider
API client (SentinelOne, Rapid7, …)	A Python parser — a `CsvProvider` subclass
Configure credentials	Nothing to configure — the file is the input
Run a sync	Upload a CSV/Excel file (a background parse job)
Normalized rows written to a collection	Same — rows written to `inventory_items`, `csv_devices`, `csv_users` or `csv_records`
Software rollup	Same `rebuild_applications` rollup for software-target sources

So a client is a Python parser and a sync is a CSV upload + parse job. Everything downstream — the data model, the software inventory, asset and user enrichment — is shared with the API sources.

The code lives in webapp/backend/app/services/csv_providers/ (base.py, providers.py, __init__.py) with generic endpoints in webapp/backend/app/api/v1/endpoints/csv_sources.py.

The `CsvProvider` base class

A provider is a subclass of CsvProvider that sets a handful of class attributes (its descriptor) and implements parse().

class CsvProvider:
    key: str = ""                       # stable id, used in URLs and as `source`
    name: str = ""                      # display name for the UI
    description: str = ""               # one-line description shown on the card
    category: str = "software"          # software | identity | access (UI grouping)
    target: str = "software"            # software | device | user | records (where it writes)
    icon: str = "FileText"              # lucide icon name used by the UI
    accepts: str = "CSV export"         # human hint about the expected file
    expected_columns: List[str] = []    # shown on the detail page as guidance
    header_signals: List[str] = []      # if set, skip preamble rows until a header row containing these

    def load(self, content: bytes) -> List[dict]: ...
    def parse(self, rows: List[dict], tenant_id: str) -> ParseResult: ...

`load()` — decode the file into rows

load() turns the uploaded bytes into a list of {header: value} dicts. You rarely override it; the base implementation:

Auto-detects CSV vs XLSX by magic bytes. If the content starts with the ZIP signature PK\x03\x04, it is treated as .xlsx and read with openpyxl (first worksheet, data_only=True); otherwise it is decoded as CSV (utf-8-sig, so a BOM is tolerated).
Skips preambles via header_signals. Some exports prepend a rules/legend block before the real header row. If a provider sets header_signals, load() scans for the first line/row that contains all of those signal strings and starts the header there. (The PTC badge export uses this.)
Trims whitespace on both headers and values.

`parse()` → `ParseResult`

parse(rows, tenant_id) maps the arbitrary export columns onto the normalized shape for the provider’s target and returns a ParseResult:

@dataclass
class ParseResult:
    records: List[dict]
    warnings: List[str] = field(default_factory=list)
    detected_shape: str = ""

records — the normalized rows to persist.
warnings — surfaced in the UI (e.g. “No rows had a recognizable application column”).
detected_shape — a short label describing what was parsed (e.g. "software", "computers", "ptc_badge"), shown on the upload drill-in.

Column helpers

Vendor exports never agree on column names or casing, so the base class provides loose matching helpers. Matching is case-, space- and punctuation-insensitive (Anti-Virus Enabled? matches isantivirusenabled).

Helper	Purpose
`pick(row, *names)`	First non-empty value among the candidate column names (matched loosely).
`pick_contains(row, *substrings)`	First value whose column name contains one of the substrings — for headers with extra text like `Supervisor's Name (First Last)`.
`has_columns(rows, *names)`	True if the first row has all the named columns — useful for shape detection.

There is also a module-level normalize_platform(os_name) that maps arbitrary OS labels onto canonical windows / macos / linux.

The four targets

target decides which collection the parsed records land in and how they are reconciled. The generic background job (_process_upload) branches on it:

`target`	Writes to	Keyed / linked by	Appears as
`software`	`inventory_items` (`source=<key>`) + software rollup	`device_id` + `application_name` + `version` (upsert)	Rows in the software inventory, attributed to the source
`device`	`csv_devices` (`source=<key>`)	`device_key` = lowercased hostname (incl. 15-char NetBIOS match)	A tab / attributes on the linked asset — never a new asset
`user`	`csv_users` (`source=<key>`)	`email` (lowercased) — links to Entra users	Enrichment on the user profile (badge access, HR details, roles)
`records`	`csv_records` (`source=<key>`)	none — reference data	Parsed records for reference / tagging

Notes on the write behaviour:

software upserts each device × application × version row and then calls rebuild_applications(tenant_id, db) so the new software rolls up alongside SentinelOne / Intune / Rapid7 / ConnectWise Automate.
device, user and records are treated as a full replacement: the job deletes the source’s prior rows for the tenant, then bulk-inserts the fresh export (a new export is the complete set).
device linking is hostname-based. The GET /{key}/device-by-name/{device_name} route matches on both the exact lowercased name and its first 15 characters (NetBIOS), because CSV device sources such as AutoElevate often carry the truncated NetBIOS hostname while Intune / SentinelOne carry the full name.
user linking is email-based (lowercased), joining to Entra users; badge sources also carry a per-site scope and role flags.

The registry

Providers are registered in __init__.py. A single list drives everything — endpoints, upload jobs, hub cards and detail pages are all generic over the registry.

_PROVIDERS: List[CsvProvider] = [
    SoftwareInventoryProvider(),
    AutoElevateProvider(),
    BadgeAccessProvider(),
    HrRosterProvider(),
]

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

def get_provider(key: str) -> Optional[CsvProvider]: ...   # by key, or None
def all_providers() -> List[CsvProvider]: ...              # all, in order

all_providers() powers GET /csv-sources/ (the hub cards).
get_provider(key) resolves the source for every per-source route.

How to add a new provider

Two steps — no endpoint, job or UI changes:

Write a CsvProvider subclass in providers.py. Set the descriptor attributes (key, name, description, category, target, icon, accepts, expected_columns, and header_signals if the export has a preamble) and implement parse(rows, tenant_id) -> ParseResult using the column helpers.
Register it by appending an instance to _PROVIDERS in __init__.py.

class MyToolProvider(CsvProvider):
    key = "my_tool"
    name = "My Tool"
    description = "What this export contributes."
    category = "software"          # software | identity | access
    target = "software"            # software | device | user | records
    icon = "Package"               # a lucide icon the UI knows
    accepts = "My Tool CSV export"
    expected_columns = ["Hostname", "Application", "Version"]

    def parse(self, rows, tenant_id):
        records, warnings = [], []
        for row in rows:
            app = self.pick(row, "application", "software", "name")
            if not app:
                continue
            records.append({
                "device_name": self.pick(row, "hostname", "device"),
                "application_name": app,
                "version": self.pick(row, "version"),
            })
        if not records:
            warnings.append("No rows had a recognizable application column.")
        return ParseResult(records=records, warnings=warnings, detected_shape="software")

The new source immediately gets a hub card, a detail page with a dropzone, upload history and purge — all from the registry.

The four shipped providers

Software Inventory CSV

Generic installed-software export (one row per device × application). Column names are matched loosely so most vendor exports work without pre-formatting.

key software_inventory · category software · target software

Field	Matched from (examples)
`application_name`	`application_name`, `application`, `software`, `name`, `product`, `display_name`
`device_name`	`device_name`, `device`, `hostname`, `computer`, `computer_name`, `machine`
`vendor`	`vendor`, `publisher`, `manufacturer`
`version`	`version`, `app_version`, `product_version`
`os_platform`	`os_platform`, `operating_system`, `os`, `platform` (normalized)
`user_name`	`user`, `user_name`, `username`, `assigned_user`

Rows without a recognizable application column are skipped; if none match, a warning is returned.

AutoElevate (CyberFOX)

The AutoElevate Computers export — a privilege-management posture source, not a software source. One row per endpoint (privilege levels, UAC status, elevation mode, admin members, AV/patch flags). It is upserted as device enrichment keyed by hostname, so it links to an existing asset and shows as an AutoElevate tab/attributes — it never creates duplicate assets or software.

key autoelevate · category identity · target device
accepts AutoElevate “Computers” CSV export

Field	Matched from (examples)
`device_name`	`name`, `Computer Name`, `device_name`
`company` / `location`	`companyName`, `locationName`
`os` / `platform`	`operatingSystem` (platform normalized)
`uac_status` / `uac_admin_level` / `uac_user_level`	`uacStatus`, `uacAdminLevel`, `uacUserLevel`
`elevation_mode` / `blocker_mode`	`elevationMode`, `autoBlockerMode`
`active_user` / `active_user_privilege`	`activeUserName`, `activeUserPrivilegeType`
`admin_members`	`adminMembersFormatted`, `adminMembers`
`agent_version` / `domain_member`	`agentVersion`, `isDomainMember`
`antivirus_enabled` / `antivirus_up_to_date`	`isAntivirusEnabled`, `isAntivirusUpToDate`
`windows_update_enabled`	`isWindowsUpdateEnabled`
`last_checkin` / `created_at`	`checkedInAtFormatted`, `createdAtFormatted`

All the extra fields land in the record’s attributes object.

Badge / Access Control

A physical access-control roster (e.g. PTC). Links badge holders — and a door-admin role — to users by email, so badge access shows on the user profile. Badge systems are per physical site, so each row is tagged with its site (the export’s Organization).

key badge_access · category access · target user
accepts PTC “Users” CSV export — the rules preamble is handled automatically via header_signals = ["Person Name", "Card No"].

Field	Matched from (examples)
`email`	`email`, `email_address`
`display_name`	`person_name`, `name`, `holder`, `cardholder`
`site`	`organization`, `site`, `location`, `company`
`card_no`	`card_no`, `card_number`, `badge_id`
`door_admin`	derived from `device_operation_role` (`2` = Administrator → `Yes`)
`person_no` / `job_title` / `phone`	`person_no`, `job_title`, `phone_no`
`on_board_date`	`on_board_date`, `start_date`

The record also carries flags.badge_admin (a boolean) and site. PTC role codes are 1 = Normal User, 2 = Administrator.

HR Roster

An HR employee roster export (CSV or .xlsx). Links employment details to users by email for owner attribution and offboarding context.

key hr_roster · category identity · target user
accepts CSV or .xlsx with roster columns (e.g. Work Email, Job Title, Department)

Field	Matched from (examples)
`email`	`work_email`, `email`, `email_address`
`display_name`	`name`/`full_name`, or `first_name` + `last_name` joined
`hr_department`	`department_description`, `department`, `dept`, `area_description`
`hr_title`	`job_title`, `title`, `role`, `position`
`manager`	`supervisors_name`, `manager`, `reports_to`, or `pick_contains("supervisor", "manager")`
`location`	`work_location_name`, `location`, `office`, `site`
`employee_id` / `company_code`	`employee_id`, `company_code`
`employment_status` / `hire_date`	`status`, `hire_date`/`start_date`

Names may be a single column or split first/last (the roster falls back to joining first_name + last_name). Employment details land in attributes.

The upload / parse job model

Every upload is recorded in the csv_uploads collection and processed by a FastAPI BackgroundTasks job (_process_upload).

Lifecycle: POST /csv-sources/{key}/upload validates the file (CSV vs Excel by extension), stores it, inserts a csv_uploads document with status = "processing", and schedules the parse job. The job runs provider.load() then provider.parse(), writes the records for the target, and updates the document.

A csv_uploads document tracks:

Field	Meaning
`provider_key`	Which source (registry key)
`filename` / `file_size`	Sanitized filename and size
`status`	`processing` → `processed` or `failed`
`row_count`	Rows read from the file
`imported_count`	Records written to the target collection
`detected_shape`	`ParseResult.detected_shape`
`warnings`	`ParseResult.warnings`
`error`	Message if the job failed
`created_at` / `processed_at`	Timestamps

Source stats and the upload drill-in

GET /csv-sources/ and GET /csv-sources/{key} return the registry descriptor plus stats: record_count (counted from the target collection), last_upload, last_status, last_filename and upload_count.
GET /csv-sources/{key} also returns the most recent uploads (history).
GET /csv-sources/{key}/uploads/{upload_id} powers the clickable drill-in. It returns the job, the target, up to 50 sample raw records from the target collection, and metrics:
- row_count, imported_count, records_in_db (live count for that upload_id), detected_shape, warnings;
- field coverage — for each field (attributes flattened in), the number and percentage of sampled rows where it is populated, sorted most-filled first. This is the fill-rate view shown on the upload modal.

Purge

DELETE /csv-sources/{key}/data removes the source’s records from its target collection (re-running the software rollup for software sources) and deletes the source’s csv_uploads history for the tenant.