Skip to content

DBT: S3 Inventory Snapshots

DBT model name: s3_inventory_snapshots

Explore dependencies/lineage: link

Description

History-keeping table of weekly S3-Inventory snapshots for trase-storage (one row per snapshot × object). Excluded prefixes are dropped at ingest. First run seeds with the latest snapshot; history accrues forward.

Details

Column Type Description
snapshot_ts `` Inventory run directory name, e.g. 2026-05-31T01-00Z (UTC).
key `` Object key within the bucket.
size_bytes `` Object size in bytes.
last_modified `` Object LastModifiedDate from the inventory (UTC).
etag `` Object ETag at snapshot time.
bucket `` Source bucket (trase-storage).

Review full report including sample errors records if they exist (link)

Test column Test name Failing rows Last test run Last status Query in Metabase
snapshot_ts dbt_expectations_expect_row_values_to_have_recent_data_s3_inventory_snapshots_strptime_snapshot_ts_Y_m_dT_H_MZ___day__9 — 2026-07-19 06:50 pass

Models

Models

  • s3_monitor_excluded_prefixes

No called script or script source not found.

"""Weekly S3-Inventory snapshots for the `trase-storage` bucket.

Incremental, history-keeping table: one row per (snapshot, object). Each weekly
run appends any inventory snapshots not already loaded, so the table accumulates
a point-in-time history of every file in the bucket.

First run seeds with only the most recent snapshot; history then accrues forward
on each subsequent run. (Old snapshots are not back-filled — that would mean
re-reading every historical full-bucket listing.)

Two layers of exclusion keep the catalog to the ~tens-of-thousands of files we
care about (vs. ~3M in the raw bucket):
  * static: any key matching a glob in the `s3_monitor_excluded_prefixes` seed;
  * dynamic: any key whose direct (non-recursive) parent prefix has more than
    `max_direct_children` file children — drops auto-generated dumps of
    thousands of sibling files.
Directory-marker keys (ending in `/`) are dropped entirely.
"""

import fnmatch

import pandas as pd


def _exclude_mask(keys: pd.Series, patterns) -> pd.Series:
    """Vectorised fnmatch: True where a key matches any exclusion glob."""
    patterns = [p for p in patterns if isinstance(p, str) and p]
    if not patterns:
        return pd.Series(False, index=keys.index)
    regex = "|".join(f"(?:{fnmatch.translate(p)})" for p in patterns)
    return keys.str.match(regex)


def _parent_prefix(keys: pd.Series) -> pd.Series:
    """Direct parent 'directory' of each key ('' for root-level files)."""
    return keys.str.rpartition("/")[0]


def _process_snapshot(df, snapshot_ts, patterns, max_direct_children, out_cols):
    """Filter + type a single snapshot's inventory frame.

    Done per-snapshot (inside the load loop) instead of on one giant
    concatenated frame, so peak memory stays ~one snapshot rather than the full
    multi-snapshot history. Both exclusion passes are per-snapshot by nature
    (static globs are row-wise; the direct-children cap is scoped to a single
    snapshot's listing), so the result is identical to filtering after a concat.

    Returns the typed `out_cols` frame, or None if nothing survives.
    """
    # Drop directory-marker keys (S3 inventory lists "foo/" with size 0).
    df = df[~df["key"].str.endswith("/")]
    # Static exclusions (seed globs).
    if patterns:
        df = df[~_exclude_mask(df["key"], patterns)]
    df = df.copy()
    # Dynamic exclusion: parents with too many direct file children.
    if not df.empty and max_direct_children > 0:
        child_counts = df.groupby(_parent_prefix(df["key"]))["key"].transform("size")
        df = df[child_counts.values <= max_direct_children].copy()
    if df.empty:
        return None

    df["snapshot_ts"] = snapshot_ts
    df["size_bytes"] = pd.to_numeric(df["size"], errors="coerce").astype("Int64")
    df["size_mb"] = (df["size_bytes"] / (1024 * 1024)).round(3)
    df["last_modified"] = pd.to_datetime(df["modified"], errors="coerce", utc=True)
    df = df.rename(columns={"hash": "etag"})
    return df[out_cols]


def model(dbt, session):
    dbt.config(
        materialized="incremental",
        unique_key=["snapshot_ts", "key"],
        tags=["s3_inventory"],
    )

    from trase.tools.aws.s3_helpers import (
        list_s3_inventory_manifests,
        read_s3_inventory_manifest,
    )

    # Tunables come from the model's +meta in dbt_project.yml — see note in
    # s3_file_schemas.
    meta = dbt.config.get("meta", {})
    max_direct_children = int(meta.get("max_direct_children", 200))

    out_cols = [
        "bucket",
        "key",
        "size_bytes",
        "size_mb",
        "last_modified",
        "etag",
        "snapshot_ts",
    ]

    # ------------------------------------------------------------------ source
    manifests = list_s3_inventory_manifests()
    if not manifests:
        raise ValueError("No S3 inventory manifests found under the inventory prefix.")

    existing_ts: set[str] = set()
    if dbt.is_incremental:
        rows = session.execute(
            f"SELECT DISTINCT snapshot_ts FROM {dbt.this}"
        ).fetchall()
        existing_ts = {r[0] for r in rows}

    todo = (
        [(ts, k) for ts, k in manifests if ts not in existing_ts]
        if existing_ts
        else manifests[-1:]  # first load: latest snapshot only
    )
    print(f"[s3_inventory_snapshots] loading {len(todo)} snapshot(s)", flush=True)

    # Load the exclusion globs once (not per snapshot). NOTE: keep
    # `dbt.ref(...).df()` on its own line — dbt's Python ref parser fails to
    # register the ref if a subscript (`[...]`) is chained directly onto it.
    excluded = dbt.ref("s3_monitor_excluded_prefixes").df()
    exclusion_patterns = excluded["prefix_glob"].to_list()

    # Read + filter + type each snapshot as we go, so only the kept
    # (post-exclusion) rows accumulate. Peak memory is ~one snapshot's raw
    # listing rather than the whole concatenated history.
    frames = []
    kept = 0
    for i, (snapshot_ts, manifest_key) in enumerate(todo, start=1):
        print(
            f"[s3_inventory_snapshots] snapshot {i}/{len(todo)}: {snapshot_ts}",
            flush=True,
        )
        df = read_s3_inventory_manifest(manifest_key, progress=True)
        if df.empty:
            continue
        df = _process_snapshot(
            df, snapshot_ts, exclusion_patterns, max_direct_children, out_cols
        )
        if df is not None and not df.empty:
            frames.append(df)
            kept += len(df)

    if not frames:
        return pd.DataFrame(columns=out_cols)

    print(
        f"[s3_inventory_snapshots] {kept:,} rows after exclusions "
        f"across {len(frames)} snapshot(s); writing",
        flush=True,
    )
    return pd.concat(frames, ignore_index=True)[out_cols]