Batch processing¶

AI Hub exposes two batch families:

• matching batches read active entities from MongoDB and calculate JOM/JPM scores;
• import batches read source data, standardize it and persist users, jobs, opportunities or trainings.

All endpoints below validate the administrator API key. They use AI_HUB_ADMIN_API_KEY when configured; otherwise they can fall back to a MongoDB key with isAdmin: true only without DATABASE_URL. Database compatibility rules still determine which operations are available.

Endpoints¶

Matching¶

Matching query parameters are client_id, data_source, environment, number_of_threads and, when relevant, pwc_staging_or_prod.

curl --get 'http://localhost:8000/admin/api/v1/matching/batches/jom/users' \
  --header 'X-API-Key: <admin-api-key>' \
  --data-urlencode 'client_id=725' \
  --data-urlencode 'data_source=smartmobility' \
  --data-urlencode 'environment=integration' \
  --data-urlencode 'number_of_threads=16'

Import and standardization¶

Common parameters are client_id, data_source, environment, deactivate_entities and number_of_threads. Each endpoint also accepts an entity-ID filter and a separate list of IDs for which skill inference should be forced. User, job and opportunity imports can clear deactivated matching scores through clean_deactivated_entities_scores; training import deliberately does not expose that parameter.

curl --request POST 'http://localhost:8000/admin/api/v1/entity-imports/trainings?client_id=725&data_source=smartmobility&environment=integration&number_of_threads=8' \
  --header 'X-API-Key: <admin-api-key>'

Execution model¶

The HTTP routes return 202 with a task_id after preflight validation. GET /admin/api/v1/operations/{task_id} reports running, success or fail while the task is retained in the current container's worker memory. Active tasks and the latest 1,000 completed outcomes per worker are retained without database persistence; evicted or lost states return 404. The application-owned task manager executes the workflow after acceptance, with bounded running and pending work. See background operations for notification recipients, capacity and shutdown behavior.

The same manager owns MongoDB skill inference, opportunity synchronization with check_only=false, JPM job-score publication, configuration normalization, PwC exports and administrative migrations. Inference keeps its existing eight-worker bound and records batch successes and failed entity IDs in the consolidated report. PostgreSQL user inference retains its separate durable reconciliation jobs.

Shared client imports and full processing¶

POST /admin/api/v1/clients defaults initialisation_import to true and imports and standardizes jobs, opportunities, trainings, then users sequentially, awaiting each batch before starting the next. These initial imports run even when JOM or JPM is disabled, without changing those feature flags or calculating matching scores. Standardization still performs configured enrichment, embeddings, persistence and email reporting. Source support remains as listed in the import table above.

POST /admin/api/v1/batch-processing/client/imports calls the same import_client_entities orchestrator in src/admin/services/client_initialization.py. POST /admin/api/v1/batch-processing/client/full awaits that orchestrator, then reuses process_one_client, also called by POST /admin/api/v1/batch-processing/client/matchings.

The shared import order is jobs → opportunities → trainings → users. Matching then runs JOM users → JOM opportunities → JPM, respecting configured matching flags. Imports and full processing require MongoDB; matching alone also supports PostgreSQL. The dedicated single-entity import endpoints keep their feature requirements. A global import error or cancellation prevents later stages; per-entity failures retain existing batch isolation. Independent matching phases continue after a reported phase failure.

Batch context¶

Each entity import batch opens one BatchProcessingContext. It reuses tenant configuration, AI configuration, the MongoDB connection, an optional department lookup and one Datalake writer. The shared import orchestrator closes each entity's context before starting the next import. Matching maintenance can share a context across its matching batches.

run_bounded_batch accepts either an async MongoDB cursor or an in-memory iterable. A fixed number of workers process entities concurrently while a queue holds at most two waiting entities per worker.

Matching input¶

Matching entities are streamed from async MongoDB cursors, so producer memory stays bounded. Embedding preparation runs once per matching type and only fills missing entity, opportunity or job embeddings. Client matching reuses the same prepared JOM context for both processing directions, including in full processing.

SQL import input¶

Smart Mobility import connectors stream SQLAlchemy results in partitions of 1,000 rows into column buffers, without an intermediate list of row dictionaries. Pandas infers types once over complete columns, preserving nullable values and joins across partition boundaries. DataFrames and the source records remain materialized for whole-source grouping and the existing deactivation workflow; this does not make the complete import constant in memory.

Smart Mobility import filters are passed to the database driver as bound parameters, including variable-length identifier lists.

An initial preparation pass validates records and counts selected entities before starting workers. A second pass feeds normalized records lazily into the bounded queue. This trades a second normalization pass for avoiding retention of all normalized users/opportunities alongside their source records, while preserving preparation failures and progress totals. PwC transformation shares the active opportunity lookup instead of copying it into multiple processes.

PostgreSQL existing-data initialization¶

Alembic does not enumerate existing business rows. PostgreSQL deployments use durable administrator runs to fill the asynchronous job tables in bounded keyset pages:

Every POST accepts an optional client_id corresponding to setting.id. Without it, all LIVE and DEMO settings are included. The response contains a durable jobId; GET /admin/api/v1/postgres-initialization/runs/{job_id} returns its steps. A step is complete after it has inserted the required child job rows. It deliberately does not wait for the entity workers to consume that backlog.

Concurrency settings¶

number_of_threads must be positive. The API returns 422 before loading SQL data when it is invalid. Values above the server ceiling are clamped and logged.

Datalake behavior¶

Every batch reuses one lazily initialized Azure Blob client. Standardized imports pass the shared writer through all four entity standardizers; matching passes it to pool/score processing. The bounded upload queue is drained when the batch context closes. When BLOB_ENDPOINT is absent or empty, all Datalake writes are silently skipped and no storage client or upload worker is started.

Unitary imports and other direct writes reuse a separate lazy Azure client per storage account and application event loop. Each direct save still awaits its upload; it does not enter the batch queue. The unified lifespan closes these clients after background processing stops. Standalone callers use connector_clients_scope() from src.common.connectors.shared_clients. An absent, empty or whitespace-only BLOB_ENDPOINT skips the write before accessing that scope or constructing any storage client, for both direct and batch paths. Upload failure reporting and one success log per stored blob are preserved.

File granularity and paths remain compatible:

• imports write one JSON object per entity below message/standardized;
• matching writes one JSON object per entity below processing/pool and processing/scores.

Single-entity HTTP and consumer flows do not provide a batch writer and retain the direct-save fallback.

Lexi generation batches¶

Lexi skill descriptions, skill levels, mission suggestions and skill suggestions use a fixed number of async workers, limited by their existing concurrency settings. Input order is preserved in the returned results, and waiting items do not each allocate a task. On an ordinary failure, the other items still run; the first exception is propagated after they finish. This can delay an error response, but leaves no unfinished sibling work behind it. Cancellation stops and awaits all workers.

Results, progress and errors¶

Every completed entity logs entity x/total; a filtered import can show ? as the total because filtering is lazy. The producer logs every 100 queued records, and the final line contains duration, successes and failures.

Failures are isolated per entity and their IDs are collected in the in-memory batch outcome, together with timing and success counts. Administrative operations normally consolidate these results in a terminal email. Client creation, the four individual entity imports, the three matching batches and client import/matching/full workflows instead isolate the report of each stage with run_client_batch_with_report, then await its email before the next stage. Imports are labelled Import; matching reports distinguish JOM users, JOM opportunities and JPM. Reports preserve succeeded, partial_failure, no_data, skipped and failed outcomes. Cancelled work receives a terminal cancellation report rather than a successful stage report.

Stage reports are merged into the overall task outcome. The full workflow always sends a terminal summary, including on success; other milestone workflows send an additional terminal report on failure or cancellation. Notification failures are isolated per recipient and do not change the processing outcome. Direct service callers outside the administrative manager retain existing per-batch emails. Batch outcomes are not persisted in a database.

The messages below are workflow results included in the completion report; they are no longer synchronous HTTP responses:

{"message": "The processing is finished."}

{"message": "No data to import."}

{"message": "The processing is not enabled for this client."}

An unknown tenant returns 404. The former nightly matching scheduler has been removed; launch batches through these APIs or the full client workflow rather than relying on nightly_batch scheduling.

Batched MongoDB vector lookups share the per-worker MONGO_VECTOR_SEARCH_CONCURRENCY budget (default 8). Their fixed worker count prevents entity batch concurrency from multiplying the number of active vector queries. Filters, indexes and scoring rules are unchanged. See configuration for failure, cancellation and standalone-script behavior.