Connectors

Code lives in app/models/connector.py, app/schemas/connector.py, app/services/connectors.py, app/services/connector_adapters/, app/services/connector_full_sync.py, app/services/connector_import.py, app/services/connector_cdr_import.py, app/core/scheduler.py, app/api/v1/routes/connectors.py.

A Connector is a tenant-configured link to one external system (ConnectorSource - currently only transatel, a closed enum: each source has its own hand-written adapter under app/services/connector_adapters/, so a new source is always a code change). It holds that system's credentials (config, validated per-source - see TransatelConnectorConfig) and, optionally, an enabled flag plus a cron schedule for automatic pulls.

Shape

connectors
  id             uuid, PK
  tenant_id      uuid | null, FK -> tenants.id, indexed; null = platform (console) template
  parent_id      uuid | null         - FK -> connectors.id; set on a tenant connector that
                                        inherits a platform connector (config = overlay only)
  name           str
  source         str                 - ConnectorSource, e.g. "transatel"
  config         jsonb               - source-specific, validated by that source's config schema
  enabled        bool                - default true; no automatic pull unless also scheduled
  schedule       str | null          - 5-field cron expression, validated with the same
                                        CronTrigger the scheduler uses
  last_synced_at timestamptz | null  - set after each scheduled run, not on-demand syncs
  created_at / updated_at / deleted_at

Inheriting a platform connector

The platform's superadmins define the global connection once, in the console (/api/v1/platform/connectors, tenant_id null): a regular connector holding the credentials and endpoints. A platform connector is a template: it can't be scheduled and every sync/import route refuses it (connectors_service.ensure_syncable).

A client tenant then creates a connector with parent_id pointing at it (GET /api/v1/connectors/templates lists what's available - identity only, never config):

POST /api/v1/connectors
{ "parent_id": "<platform connector id>",
  "config": { "subscriber_group_filter": ["CELTIC", "CELTIC_DEMO"],
              "account_label": "Celtic",
              "group_account_labels": { "CELTIC_DEMO": "Celtic DEMO" } } }

• source (and name, if omitted) come from the parent; config is only the tenant's overlay and is validated against the source's tenant-overridable keys (overridable_fields() in app/schemas/connector.py: subscriber_group_filter, account_label, group_account_labels, cdr_lookback_days). Credentials/endpoints are rejected - they only live in the platform connector.
• The effective config is parent.config | overlay (connector_config.effective_config), resolved at read time by every adapter, sync and import, so rotating a credential in the console applies to all inheriting tenants immediately. ConnectorRead.config is the effective config (secrets masked), overrides is the overlay alone, inherited/parent_id identify the link.
• PATCH of config on an inheriting connector merges into the overlay; a null value removes that override. enabled/schedule are always the tenant's own.
• Account label: after each pull, groups attach to the Account whose company name equals group_account_labels[group], else account_label (the account is created if missing). Only groups with no account yet are attached; an already-attached group is never re-pointed.
• One inheriting connector per platform connector and tenant (409 otherwise); a platform connector with live inheriting children can't be deleted. A tenant can still create a standalone connector with its own credentials instead of inheriting.
• Shared SFTP CDR drops aren't split per tenant yet: CDR ingestion of an inherited connector sees the whole account's files.

Automatic sync

app/core/scheduler.py runs one in-process APScheduler AsyncIOScheduler per API process, with one cron job per enabled+scheduled tenant connector (sync_job, keyed by connector id). connectors_service's create/update/delete call sync_job/unschedule for the local process, and reload_connector_jobs re-reads the schedules from the DB every 60 seconds so a write made through another process reaches this one too. start() (called from app/main.py's lifespan) reaps orphaned events, loads the connector jobs and registers the housekeeping jobs.

Multiple processes are tolerated, not coordinated by an external queue: a sync takes a per-connector lock (events.sync_lock_key) and a tick another process already ran is skipped (ConflictError); housekeeping jobs run under a job:<name> lock. A missed run (coalesce=True, misfire_grace_time=1s) is skipped, not queued up as a backlog.

Each firing runs app/services/connector_full_sync.py's run_full_sync: refresh plans (plan_sync; platform plans are skipped, see Plans), pull rate plans (rate_sync), pull subscribers grouped by Transatel group (creating a SubscriberGroup per distinct group value, via connector_import.import_subscribers_by_group), pull new Rated CDR files off SFTP (connector_cdr_import.sync_cdr_files - see "CDR ingestion" below), backfill plan_id across all of the tenant's subscribers by profile name (subscriber_plan_matching), then create missing Customers from subscriber emails (customer_sync). The subscriber, CDR and customer stages are isolated - a failure is reported on the result (*_error), not raised. The whole run is wrapped in an Event row (trigger=scheduled) - see Events.

CDR ingestion

Transatel delivers rated usage as gzipped CSV files dropped on an SFTP server (sftp://<sftp_username>:<sftp_password>@<sftp_host>/<sftp_remote_dir>/<seq>_RatedCDR_ <yyyyMMddHHmmss>_<n>.csv.gz - all of sftp_host/sftp_username/sftp_password/ sftp_remote_dir/cdr_lookback_days live on TransatelConnectorConfig, same posture as every other credential here). app/services/connector_adapters/transatel_sftp.py does the blocking paramiko/gzip/CSV mechanics (run off the event loop via asyncio.to_thread, since paramiko has no async API); TransatelAdapter.list_cdr_files/fetch_cdr_file_rows wrap it.

connector_cdr_import.sync_cdr_files orchestrates: list files generated within cdr_lookback_days, skip any whose filename is already recorded in cdr_ingest_files for this connector (so a re-run only touches files that actually just landed), then for each new file download + ungzip + parse and upsert its rows into cdr_records (ON CONFLICT (tenant_id, notification_id) DO NOTHING, where notification_id is <filename>:<row index> - see app/models/cdr_record.py). Known Rated CDR columns are mapped onto CdrRecord's typed fields (msisdn, event_timestamp, usage_val, ... - _TEXT_COLUMNS in connector_cdr_import.py), other non-empty columns go to extra, and the exact raw row is kept under row_metadata["_raw"]; an unparseable value becomes null rather than failing the file. Promoting a CdrRecord into a priced Line is not done.

A connector with no sftp_username/sftp_password configured is skipped, not treated as a failure - most connectors only need the HTTP-side sync above.

API

Method	Path	Notes
POST	/api/v1/connectors	Create (config validated against source at save time)
GET	/api/v1/connectors	Paginated list, ?source=transatel
GET	/api/v1/connectors/sources	Supported sources with their config/import form specs
GET	/api/v1/connectors/templates	Platform (console) connectors this tenant can inherit (id/name/source only)
GET	/api/v1/connectors/{id}
PATCH	/api/v1/connectors/{id}	source is immutable; explicit schedule: null clears it, omitted leaves it unchanged
DELETE	/api/v1/connectors/{id}	Soft delete; unschedules any cron job; 409 while inheriting children exist
POST	/api/v1/connectors/{id}/sync?kind=customers|subscribers|usage	On-demand pull of one data kind - not wrapped in an Event, not part of the full-sync pipeline. kind=usage runs connector_cdr_import.sync_cdr_files directly (not adapter.pull_usage(), which stays an unimplemented stub - see "CDR ingestion" above); customers/subscribers go through the source's adapter
POST	/api/v1/connectors/{id}/sync-all	Runs the same full pipeline a scheduled job runs, on demand. Tracked as an Event with trigger=manual
POST	/api/v1/connectors/{id}/import/subscribers	Manual, operator-triggered pull of subscriber rows into the CSV import engine's mapping/match/upsert pipeline (see Customer/Subscriber CSV Import), with an explicit group_id/filters. Deliberately not part of the automatic schedule - a cron job has no source for those operator-supplied inputs

Both sync routes return 501 when the adapter doesn't implement the kind; the subscriber import returns 502 when the upstream API refuses the call.

What's deliberately not built

• No source besides Transatel - _CONFIG_SCHEMAS (app/schemas/connector.py) is the single place a new source's config shape would register.
• POST /{id}/sync (single data kind) is not tracked as an Event; only the full pipeline (sync-all / the scheduled job) is.