Customer/Subscriber CSV Import

Code lives in app/models/import_mapping.py, app/models/import_run.py, app/schemas/import_mapping.py, app/schemas/import_run.py, app/services/import_mapping_engine.py, app/services/import_engine.py, app/services/import_mappings.py, app/services/import_runs.py, app/services/csv_parser.py, app/api/v1/routes/import_mappings.py, app/api/v1/routes/import_runs.py.

This is a reusable mapping/upsert/reporting core, not a single-shot importer, so a plain tenant CSV upload and an API-based source (Transatel) share the same engine.

What this isn't

• Not just the CSV-upload path anymore. app/services/connector_import.py builds a MappingContext from a connector's fetched rows (Transatel fetch_subscribers/search_lines, app/services/connector_adapters/transatel.py) and runs them through the same mapping/match/upsert/report engine described on this page, via POST /connectors/{id}/import/subscribers and the automatic per-connector sync (import_subscribers_by_group). See Connectors for that trigger and its grouping behavior; this page still covers the engine itself and the CSV-upload trigger (POST /import-mappings/{id}/runs), which the connector path reuses unmodified.
• A single import run is still not a background job - it executes synchronously inside the request that triggers it (upload, or the connector-import call), bounded by MAX_ROWS/MAX_FILE_SIZE_BYTES (app/services/csv_parser.py) for the CSV path. That said, app/core/scheduler.py runs an in-process APScheduler cron job per connector to fire its automatic full sync (which may include a subscriber import) - see Connectors. There is no generic background-job system.

Shape

import_mappings
  id            uuid, PK
  tenant_id     uuid, FK -> tenants.id, indexed
  name          str                    - user-facing label
  target        str                    - "account" | "subscriber"
  source_type   str                    - "csv" (the only `ImportSourceType` member)
  config        jsonb                  - the mapping DSL, see below
  created_at / updated_at / deleted_at  - soft-deleted, mutable catalog row

import_runs
  id                      uuid, PK
  tenant_id               uuid, FK -> tenants.id, indexed
  mapping_id              uuid | null, FK -> import_mappings.id, indexed - null for a connector-
                                          triggered run (its mapping is fixed in code,
                                          `connector_import._TRANSATEL_MAPPING_CONFIGS`)
  source_key              str          - the `external_source` identity matching scopes on:
                                          `str(mapping_id)` for CSV, e.g. "transatel:sim_search"
                                          for a connector run
  source_type             str          - denormalized from the mapping at run-creation time
  status                  str          - "pending" | "running" | "succeeded" | "failed" | "partial"
  dry_run                 bool
  rows_total/created/updated/failed/skipped  int
  row_results             jsonb        - list[{row_index, external_key, action, error}], capped
                                          at ROW_RESULTS_CAP (1000) - beyond the cap only failed
                                          rows stay inline and rows_results_truncated is set
  rows_results_truncated  bool
  started_at / finished_at / created_at  - no deleted_at: immutable audit record

Account/Subscriber each gained external_source/external_id (nullable strings) - a cross-reference to the record's origin in an upstream system. Deliberately a plain composite index, not a DB-level unique constraint, on (tenant_id, external_source, external_id): both entities are revisioned, and permanent_change's split path can leave two live rows sharing the same logical id (and therefore the same external_id, which isn't part of either *Update schema) - a hard uniqueness constraint spanning every revision would make that existing, legitimate state raise IntegrityError. "One match per external_id" is instead enforced by the match engine scoping its query to the currently-active revision (from_date <= now < to_date), matching how get_account/get_subscriber already resolve "the" row for an id.

The mapping DSL

import_mappings.config (validated by ImportMappingConfig, app/schemas/import_mapping.py):

{
  "version": 1,
  "delimiter": ",",
  "fields": [
    { "target": "first_name", "source": "FirstName" },
    { "target": "last_name", "source": "LastName", "required": true },
    { "target": "extra.loyalty_tier", "source": "LoyaltyTier" }
  ],
  "primary_key": { "match_on": "external_id", "source": "CustomerID" },
  "upsert_policy": "upsert"
}

fields[] is rename-only: mapped[target] = row[source], verbatim, no per-field transformation. An earlier column-operations catalog (concat/split/const/date/int/float/bool/lookup/regex/trim/ upper/lower) was built and then removed - see FieldSpec's own docstring (app/schemas/import_mapping.py). None of the fixed ops handled real-world CSV messiness well (a live Transatel export's name column defeated a naive split), and a JS/expression evaluator was considered and rejected for the same reason apply_row has no eval today: running tenant-authored code server-side in a multi-tenant billing backend is a real attack surface, not a small addition. A safe, non-eval expression mechanism (a constrained formula grammar, not general-purpose code) is the planned replacement once scoped - until then, a mapping does column selection only. A target of "extra.<name>" (any name not matching a real Account/Subscriber field) is the one exception: it's folded into the record's extra JSONB bucket instead of requiring a real column to exist for every CSV field (see Shape, above). Every field is attempted regardless of an earlier field's failure - a required field's emptiness doesn't stop sibling fields from still populating the row. Two fields can never share one output target - rejected at mapping-save time (ImportMappingConfig's validator), not silently last-one-wins.

Independent of fields[]/mapping config, every imported row's entire raw source dict (before any mapping is applied) is stamped into the record's separate metadata JSONB bucket under metadata["raw_import"], on every create and update - an unconditional audit trail, not something a tenant configures. See app/services/import_engine.py:run_import and Account.metadata_/Subscriber.metadata_.

Since fields[] can't transform a value (only rename it), a connector adapter that needs to translate its own status vocabulary onto Subscriber.status (SubscriberStatus, app/schemas/enums.py) does so before the row ever reaches the import engine, adding an already- translated mapped_status key alongside the untouched raw one - see app/services/connector_adapters/transatel.py's _flatten_subscriber_record/_flatten_line_record and app/services/subscriber_status_mapping.py. A mapping targeting subscriber.status should source from mapped_status, not the raw status/lineStatus field.

primary_key.match_on (ImportMatchOn): external_id, email (Account only), subscriber_id / account_id (the row already carries our own id), none (always create, no match attempted).

upsert_policy (ImportUpsertPolicy): upsert (default: update on match, create otherwise), insert_only (skip on match), update_only (skip when no match), create_new_on_conflict (always create, even on match - an explicit duplicate), error_on_conflict (fail the row on match).

Match/upsert engine

app/services/import_engine.py. Decision: batch the Match, keep Upsert per-row. Given a run can have thousands of rows, a naive per-row SELECT would be that many round trips:

1. Phase A (pure, no DB) - every raw row runs through apply_row (app/services/import_mapping_engine.py) plus a lenient schema check (AccountCreate for Account - it already has no required fields; SubscriberBatchItem, not SubscriberCreate, for Subscriber, since a row that's actually going to update an existing subscriber shouldn't be forced to also supply account_id/plan_id). The row's raw mapped dict (only the fields the CSV actually populated) is what's carried forward, not the schema's defaulted round-trip - otherwise every update would silently reset every field the mapping doesn't cover (e.g. detailed_invoice back to its schema default) instead of leaving it untouched.
2. Phase B - one batched WHERE tenant_id=... AND <column> IN (...) AND <active-revision predicate> IN (...) query for the whole run (chunked at 5000 keys), building an in-memory lookup dict. external_id matches are further scoped by external_source == str(mapping_id) - a mapping is the source, in this system's terms, so two different mappings' ids never collide even if spelled the same.
3. Phase C - per row, apply upsert_policy against whether Phase B found a match, then call the existing, unmodified create_account/update_account/create_subscriber/ update_subscriber service functions - each already does its own per-item db.add/commit/refresh, so writes stay per-row/per-commit, the same "no savepoints, per-item isolation" posture batch_upsert_subscribers already established. external_source/ external_id are stamped onto the ORM object directly after a create call returns, rather than threaded through AccountCreate/SubscriberCreate - those schemas stay untouched by this domain.

Reporting and dry runs

• Per-row visibility in the style of SubscriberBatchResult/BatchUpsertAction - ImportRowResult widens it with action=skipped (a policy-driven skip) and external_key (cross-referencing a failed row back to the source file).
• Dry-run runs the identical code path, short-circuiting right before the actual DB write - not a parallel preview implementation that could drift from what a real run does. An ImportRun row is still persisted for a dry run (dry_run=true is the only marker), so a preview is revisitable via GET /import-runs/{id}, not a throwaway response.

Limitations

• No write-batching. Only the Match lookup is batched; Upsert stays per-row, reusing create_account/update_account/create_subscriber/update_subscriber unmodified rather than duplicating their validation/business rules or dropping per-row isolation. Flagged as a future optimization if a real large import proves per-row commits too slow - not a documented product requirement today.
• No streaming CSV parser. parse_csv (app/services/csv_parser.py) buffers the whole file into memory before the pipeline runs - a true streaming pipeline would force the batched Match back to per-row, defeating the point of batching it. MAX_FILE_SIZE_BYTES (20MB) /MAX_ROWS (50,000) bound a single request's memory/time; they're this implementation's own limits, not a documented product ceiling.
• No configurable encoding/delimiter sniffing beyond a fixed utf-8-sig decode (BOM-safe) and an explicit per-mapping delimiter.
• No DB-level uniqueness on external_id - see the Shape section above.
• No mapping versioning/edit-history. An edited ImportMapping just changes going forward; nothing reads what an old version looked like for ImportRuns already executed - the run's own row_results is the durable record of what actually happened, independent of later mapping edits.

API

Method	Path	Notes
POST	/api/v1/import-mappings	Create a mapping (config only, no file yet)
GET	/api/v1/import-mappings	Paginated list, ?target=account|subscriber
GET	/api/v1/import-mappings/{id}
PATCH	/api/v1/import-mappings/{id}	name/config only - target/source_type are immutable after create
DELETE	/api/v1/import-mappings/{id}	Soft delete - existing ImportRuns are untouched
POST	/api/v1/import-mappings/{id}/runs	Multipart file upload, ?dry_run=true for a preview. Rejects (400) before creating a run row if the file itself can't be parsed (oversized, too many rows, not valid CSV) - that's "there is no data", not "some rows failed"
GET	/api/v1/import-mappings/{id}/runs	Paginated run history for one mapping
GET	/api/v1/import-runs/{id}	One run's full detail, including row_results