Component Reference

Advanced configuration for ai_core component classes. See Code Structure for the component overview and basic YAML format.

Raw Tier Schema

Every raw asset stores exactly five columns (produced by make_raw_row() in ai_core/schema.py):

Column	Type	Content
_partition_date	date	Partition date the fetch ran for
_fetched_at	datetime	UTC timestamp of the fetch
_request_url	str	URL called
_request_params	str	JSON-encoded query params
_response_body	str	Raw JSON response string

TransformedEntityComponent reads _response_body and passes it to the extractor in the source registry.

ai_core Utilities

packages/ai_core/src/ai_core/http/ and packages/ai_core/src/ai_core/patterns.py provide reusable helpers and dataclasses for source modules:

• @retry_with_exponential_backoff — decorator that retries on transient HTTP errors with jitter. On retry exhaustion the underlying exception is re-raised (it does not return None), so callers can derive a final status code instead of silently swallowing the failure. Use on any fetch_* function that calls an external API.
• call_with_exponential_backoff(fn, *, max_retries, backoff_seconds, exceptions, op_name=None) — imperative form of the same; use when retry params come from variables. Caller passes a no-arg callable (typically lambda or functools.partial).
• http_get_with_retry(url, ...) / http_post_with_retry(url, ...) — single-call HTTP GET/POST with exponential backoff on 429 / 5xx + connection errors. Wraps httpx.get/post with call_with_exponential_backoff and TransientHTTPError.raise_if_transient.
• paginated_get(url, params, ...) / paginated_post(url, body, ...) — fetch all pages of a paginated REST endpoint concurrently. Both return a flat list of (response_content, request_params) tuples — one per fetched page, in arbitrary order when max_workers > 1. Pass the content directly to make_raw_row to populate the _response_body Iceberg column without re-encoding.
• paginated_get_events(url, params, ...) — event-native form of paginated_get from ai_core/http/paginated.py. Emits HttpRequestEvent records for response and exception events, including retry attempts, attempt count, and request duration, so callers that need request transparency can inspect failures without parsing retry logs. The regular paginated_get helper remains the compatibility wrapper that yields only successful response pages.
• TransientHTTPError(message, *, status_code=None) — exception raised by HTTP helpers; carries an optional status_code so callers can derive the final response status after retry exhaustion. Class-level helpers: TransientHTTPError.is_transient(status_code) (predicate for 429 / 5xx) and TransientHTTPError.raise_if_transient(resp) (raise on transient status, return resp otherwise).
• extract_status_code(exc, resp=None) — single-source helper that pulls a status code from TransientHTTPError, httpx.HTTPStatusError, or a fallback response object.
• FetchAttempt(ok, status_code, context) — dataclass recording a single HTTP attempt. Source modules emit one per request to feed the fetch_attempts asset check; context is a small dict naming what the request was for (e.g. {"page": 4, "vehicle_id": "v123"}).
• RawFetchResult(df, attempts) — dataclass returned by fetch() and fetch_with_upstream() on RawSourceComponent. Bundles the raw pl.DataFrame and the FetchAttempt records that produced it.
• AkamaiSession — stateful curl_cffi session for sites protected by Akamai Bot Manager. Handles TLS fingerprinting, Akamai-specific cookie management via seed_url warmup, and automatic session rotation on bot-detection responses. Used by inventory source modules; instantiate and use as a context manager.
• browser_session(*, seed_url=None, impersonate="chrome", ...) — context manager yielding a curl_cffi session with browser TLS impersonation. Use for sites with TLS-fingerprinting bot protection (Akamai, Cloudflare, PerimeterX); seed_url triggers a warmup GET to set bot-manager session cookies before the first real call.
• parallel_call(tasks, *, max_workers, ...) — fan out a list of zero-arg callables across a thread pool. Failures captured as exceptions in the result list (per-input). Supports Hystrix-style circuit-breaker abort via failure_ratio + min_samples, or abort_on_any_failure=True for "drop on first failure" semantics; aborted runs raise ParallelCallAborted carrying the partial results.
• make_raw_row_frame(partition_date, request_url, request_params_list, response_body_list, *, fetched_at) — builds a pl.DataFrame matching RAW_ROW_SCHEMA from columnar input lists. Prefer this over [make_raw_row(...) for ...] + pl.from_dicts(rows, ...) when a fetcher produces many rows: constants are broadcast into a Series of length n instead of repeated in n separate Python dicts, and the transient list[dict] intermediate that previously coexisted with the resulting DataFrame is gone. See make_raw_row for single-row construction.

Source modules should use these rather than implementing their own retry, pagination, or instrumentation logic.

For memory and wall-time considerations, including measured wins from the patterns we adopted and a list of deferred enhancements, see Performance Notes.

Source Overrides

PPM_SOURCE_OVERRIDES is an optional env var pointing to a YAML file that substitutes local fixture data for live API calls. This allows full pipeline materializations without network access, and is the preferred approach for integration testing. See ai_core/source_overrides.py for the override resolution logic.

RawSourceComponent

Fetch contract

fetch() and fetch_with_upstream() return a single RawFetchResult. The df field may be a pl.DataFrame (in-memory) or a pl.LazyFrame (streaming, backed by a temp IPC file). Use drain_fetch_results from ai_core.http when a fetcher produces data in natural per-resource batches — it drains a generator of RawFetchResult chunks into a single RawFetchResult with a scanning LazyFrame, avoiding peak memory spikes while keeping the component interface simple.

fetch_attempts instrumentation

The raw-tier fetch_attempts asset check (severity WARN, non-blocking) reports per-request outcomes in its metadata so non-zero HTTP failures show up in the Dagster UI without parsing logs. To feed it, the source module's fetch_* function should:

1. Catch TransientHTTPError and httpx.HTTPError per request.
2. Build a FetchAttempt for each request — set ok=True on success, or ok=False with status_code derived via extract_status_code(exc, resp) on failure. Populate context with whatever fields identify the request (e.g. {"target": vehicle_id, "page": page_idx}).
3. Return RawFetchResult(df=df, attempts=attempts) from the fetcher, and have the component's fetch() return that value unchanged.

OEMs that have not been instrumented continue to work — the component wraps a bare pl.DataFrame return into a RawFetchResult automatically and the asset check reports tracking=uninstrumented in its metadata so operators can tell instrumented and uninstrumented sources apart at a glance.

The check's metadata also includes failure_patterns — failures grouped by (status_code, set of context keys) and capped at MAX_PATTERNS_IN_METADATA = 50 so the JSON payload stays small enough for the Dagster UI. Each pattern surfaces common_fields (constant context across the cluster — the actionable signal) and variations (per-attempt values for the keys that differ).

Concurrency

Set concurrency: N in YAML when a raw component makes many parallel requests (e.g. per-vehicle detail fetches). The base class uses this to cap in-flight requests.

Upstream Source Dependencies

When a raw fetch requires data from another raw asset (e.g. dealer IDs must be fetched first), override fetch_with_upstream() instead of fetch() in the RawSourceComponent subclass. The upstream asset key is declared via the upstream_dep attribute in YAML as a full asset key path (e.g., "audi/raw/scs_search"). The base class wires the dependency automatically.

BigQuery-backed Sources

When a raw component reads from BigQuery instead of an HTTP API, it uses a BigQueryResource (from dagster-gcp) registered in the OEM project's Definitions under the key BIG_QUERY_RESOURCE_KEY (exported from ai_core.components.bigquery).

The resource is registered once per OEM project and supplies the GCP project and credentials:

from dagster_gcp import BigQueryResource
from ai_core.components.bigquery import BIG_QUERY_RESOURCE_KEY

dg.Definitions(resources={
    BIG_QUERY_RESOURCE_KEY: BigQueryResource(
        gcp_credentials=dg.EnvVar("GCP_CREDENTIALS"),
        project=os.environ.get("<OEM>_GCP_PROJECT", "<default-project>"),
    )
})

Authentication uses Application Default Credentials when gcp_credentials is not set — run gcloud auth application-default login for local development.

Tags

RawSourceComponent accepts a tags: dict[str, str] attribute. The component merges this with {"oem": oem} on every asset it produces. OEMs with multiple brands use this to attach a brand tag:

tags:
  brand: jeep

Each brand gets its own raw/transformed components; the consolidated tier unions them by listing all brand-scoped transformed asset keys as separate source_groups entries.

TransformedEntityComponent

Tiebreak

Selects a winning row when duplicate records conflict on non-aggregated columns:

tiebreak:
  column: date_offer   # field to compare when resolving conflicts
  strategy: max        # or min

Aggregations

Collects distinct values from duplicate rows into a list column:

aggregations:
  - output_name: vsapi_classifications
    collect_distinct_from: vsapi_classification

ConsolidatedComponent

See the ConsolidatedComponent reference for full documentation, including:

• Source Groups — history_days, observation_date_col, filter_window_boundary, gap fill
• Merge Spec — match_on tiers, coalesce policies
• Foreign Keys — single-value joins, left_col>ref_col syntax, list-mode dot-path joins

Cross-source overlap

The inventory entity spec accepts an optional overlap_check. Where cross_source_discrepancies compares field values for keys that already match, overlap_check compares set membership — verifying that entities expected in two feeds actually appear in both. It is the invariant that catches a feed gap where a vehicle silently drops out of one source.

The check is emitted as a multi_asset_check over the transformed assets of the paired source groups (not the consolidated asset), so it runs as soon as those transformed assets materialize. Each pair names two source groups by their SourceGroup.name; groups not named in any pair (e.g. a sales-history feed) are ignored, which is how the check is scoped to feeds expected to overlap.

inventory:
  source_groups:
    - { name: dc_active, source: inventory_from_dealerconnect_bq_active, priority: 1 }
    - { name: public,    source: inventory_api_inventory,                priority: 2 }
    - { name: dc_sold,   source: inventory_from_dealerconnect_bq_sold,   priority: 3 }
  overlap_check:
    match_on: [ vin ]            # join key identifying an entity across both feeds
    blocking: true              # ERROR + skip consolidated on failure; false = WARN only
    breakdown_by: [ brand, model_code ]   # group the "missing from other feed" counts
    pairs:
      - left: public
        right: dc_active        # dc_sold is omitted → not compared
        left_coverage:  { min: 0.85 }   # frac of public VINs present in dc_active
        right_coverage: { min: 0.0 }    # frac of dc_active VINs present in public
        right_filter:                   # restrict a side to its comparable subset
          - { column: order_type, op: not_in, values: [in_transit, in_production] }

Coverage is directional: left_coverage = |left ∩ right| / |left|, and symmetrically for the right. Each side's coverage must fall within its {min, max} band or the check fails for that side's transformed asset. An empty side scores 0.0 coverage so a wholly missing feed fails rather than passing vacuously. *_filter predicates compare the column as a string; not_in/ne retain null rows, in/eq drop them.

Tuning the directions. The two directions catch different gaps and have different sensitivities:

• left_coverage (public → private) catches a private active-feed gap — vehicles the public scrape still sees that the private feed dropped. It needs no filter.
• right_coverage (private → public) is confounded by units a private feed legitimately carries but a public scrape never lists (in-transit / in-production, often 25–30% of allocation). Give it a meaningful floor only after a right_filter scopes the private side to retail-ready, on-lot units.

Blocking is a same-run gate. A failing blocking check raises DagsterAssetCheckFailedError and skips assets downstream of the checked transformed assets in the same run — so in a full pipeline materialization (raw → transformed → consolidated) it short-circuits consolidated/inventory. It does not gate a selective re-materialization of only consolidated/inventory, which reads transformed from Iceberg without re-running the check. Treat blocking as a strong guard on the daily pipeline, not an airtight cross-run invariant. Roll a new check out with blocking: false to observe coverage in the UI, calibrate the bands, then flip it on.