Nine customers, one loaderfilesystem-common and the landing-bucket bronze buildout

def _reader(
    resolve_fmt, extractions, capture_s3_modified: bool,
    sink: "RejectSink | None" = None, asserts: Optional[dict] = None,
):
    """The one dlt transformer body: per file, resolve a :class:`CsvFormat` (fixed
    or sniffed), decode through the tested core, and optionally attach the S3
    ``modification_date`` as ``_s3_modified``. When a ``sink`` is given, quarantined
    rows are recorded into it (per file) instead of vanishing silently. ``asserts``
    is the **fixed-mode** drift guard (the sniff path asserts inside ``resolve_fmt``)."""

    @dlt.transformer
    def _read(items):
        for item in items:
            data = item.read_bytes()
            if asserts:
                check_asserts(data[:65536], asserts)
            path = _path_of(item)
            rejects: list[str] | None = [] if sink is not None else None
            table = process_file(data, path, resolve_fmt(data), extractions, rejects=rejects)
            if sink is not None and rejects:
                sink.record(path, rejects)
            if capture_s3_modified:
                table = apply_file_metadata(
                    table, {"modification_date": item.get("modification_date")}, _S3_MODIFIED
                )
            yield table

    return _read

def decode(data: bytes, fmt: CsvFormat, *, rejects: list | None = None):
    """Parse ``data`` into a pyarrow Table of all-VARCHAR columns named from the
    header. ``fmt`` carries the delimiter (possibly multi-byte), quote char (or
    None), source encoding, NUL-scrub flag, and ragged-row policy. For
    ``ragged="quarantine"``, when a ``rejects`` list is given the raw rejected lines
    are appended to it (visible, not silently dropped)."""
    import os
    import tempfile

    import duckdb

    data = _to_utf8(data, fmt.encoding)
    if fmt.strip_nul and b"\x00" in data:
        data = data.replace(b"\x00", b"")

    if fmt.ragged == "reconstruct":
        text = data.decode("utf-8")
        header = text.split("\n", 1)[0].rstrip("\r")
        width = header.count(fmt.delimiter) + 1
        data = _reconstruct(text, fmt.delimiter, width).encode("utf-8")
    elif fmt.ragged == "quarantine":
        data, rejected = _quarantine_filter(data, fmt)
        if rejects is not None:
            rejects.extend(rejected)

    # "error": no flags -> DuckDB raises on a ragged row. "reconstruct"/"quarantine":
    # the data was made rectangular above, so a plain parse suffices.
    kwargs = _csv_kwargs(fmt)

    tmp = tempfile.NamedTemporaryFile(suffix=".csv", delete=False)
    try:
        tmp.write(data)
        tmp.close()
        conn = duckdb.connect()
        try:
            return conn.read_csv(tmp.name, **kwargs).to_arrow_table()
        finally:
            conn.close()
    finally:
        os.unlink(tmp.name)

_SELECTIONS = ("all", "latest_partition", "incremental")
_SEMANTICS = ("snapshot", "delta", "rolling")
# How a deletion in the source becomes a tombstone in the frontier. Orthogonal to
# ``semantics`` (the physical shape): a full *snapshot* can still be ``in_band`` if
# rows age out of a window rather than being deleted (SSR), so absence is unreliable.
#   - ``none``    don't derive tombstones (default — never manufacture deletes).
#   - ``absence`` present-in-an-earlier-snapshot-but-absent-from-the-latest = deleted.
#                 ONLY sound for a *bounded* full snapshot (a true master/dimension
#                 export); the frontier must gate it behind a completeness fence.
#   - ``in_band`` the source carries the delete itself (a flag/status column, named
#                 by ``soft_delete``); absence means "aged out", NOT deleted.
_DELETES = ("none", "absence", "in_band")


@dataclass
class TableSpec:
    """One logical table. ``file_mask`` is a glob; the remaining fields are the
    structured metadata that silver consumes."""

    name: str
    file_mask: str
    selection: str = "incremental"
    selection_key: str = "landed_at"
    pk: list[str] = field(default_factory=list)
    semantics: Optional[str] = None
    soft_delete: Optional[str] = None
    deletes: str = "none"
    provenance: list[Extraction] = field(default_factory=list)

@dataclass(frozen=True)
class CsvFormat:
    """How to turn a delimited file's bytes into rows. Bronze always lands every
    column as VARCHAR (typing is a silver concern), so there is deliberately no
    column/dtype machinery here — only the bytes-to-rows facts.

    delimiter: str = ","
    quote: Optional[str] = '"'
    encoding: str = "utf-8"
    header: bool = True
    strip_nul: bool = False
    ragged: RaggedPolicy = "error"

    def override(self, **fields) -> "CsvFormat":
        """Return a copy with the provided fields replaced. **Omission means
        inherit** — a per-table override passes only the keys it changes — while a
        passed value (including ``None`` for ``quote``, i.e. "no quoting", or
        ``False`` for a bool) is applied. This avoids overloading ``None`` to mean
        both "inherit" and "no quoting", which a quote-less feed genuinely needs."""
        known = {f.name for f in dataclasses.fields(self)}
        unknown = set(fields) - known
        if unknown:
            raise ValueError(f"unknown CsvFormat field(s): {sorted(unknown)}")
        return replace(self, **fields)

class RejectSink:
    """Accumulates quarantined (malformed) rows across a load so the component can
    surface them as Dagster metadata, rather than ``ragged="quarantine"`` dropping
    them with no trace. One sink per asset run; the reader records into it per file.

    ``total``/``files`` are the row and file counts; ``sample`` holds up to
    ``sample_limit`` ``"<path>: <raw line>"`` strings for a glance at what was
    rejected (bounded so a pathological file can't blow up run metadata)."""

    def __init__(self, sample_limit: int = 20):
        self.sample_limit = sample_limit
        self.total = 0
        self.files = 0
        self.sample: list[str] = []

    def record(self, path: str, rejected: list[str]) -> None:
        if not rejected:
            return
        self.total += len(rejected)
        self.files += 1
        room = self.sample_limit - len(self.sample)
        if room > 0:
            self.sample.extend(f"{path}: {line}" for line in rejected[:room])

    def _validate(self) -> None:
        """Fail at definition-build time on config the run path would misread.

        The run path dispatches on string equality (``selection == "incremental"``
        decides append vs replace), so an unvalidated typo like ``incrmental`` would
        silently flip an append-only raw log to ``replace``. The enums (``selection``,
        ``semantics``/``deletes``, ``in_band`` requires ``soft_delete``, duplicate
        table names) are enforced by mapping the config through the same
        ``build_source_spec`` seam the core tests exercise, so YAML and
        :class:`~filesystem_common.core.spec.TableSpec` can't drift apart silently."""
        known_facts = {f.name for f in dataclass_fields(CsvFormat)}
        unknown = set(self.format.asserts) - known_facts
        if unknown:
            raise ValueError(
                f"format.asserts keys {sorted(unknown)} are not CsvFormat facts "
                f"({sorted(known_facts)})"
            )
        self.format.to_core()  # validates ragged + non-empty delimiter

        creds = self.credentials or None
        mask = self._resolve_mask(table_config, creds)
        disposition = "append" if table_config.selection == "incremental" else "replace"
        sink = RejectSink()
        # dlt fully extracts (running our reader, populating the sink) before the
        # resource yields its MaterializeResult, so the sink is complete by then —
        # merge the quarantine counts onto the result instead of dropping them.
        for result in dlt_pipeline_resource.run(
            context=context,
            dlt_source=self._make_source(table_config, mask=mask, sink=sink),
            dlt_pipeline=self._make_pipeline(table_config.name),
            loader_file_format="parquet",
            write_disposition=disposition,
        ):
            if sink.total and isinstance(result, dg.MaterializeResult):
                result = result._replace(
                    metadata={**(result.metadata or {}), **_reject_metadata(sink)}
                )
            yield result
        if sink.total:
            context.log.warning(
                "%s: quarantined %d malformed row(s) across %d file(s); "
                "see asset metadata for a sample",
                table_config.name, sink.total, sink.files,
            )

def compile_filters(filters: dict[str, list[dict]]) -> dict[str, str]:
    result = {}
    for entity, entries in filters.items():
        if not entries:
            continue
        parts = [_compile_entry(e) for e in entries]
        if len(parts) == 1:
            result[entity] = parts[0]
        else:
            result[entity] = "(" + " OR ".join(parts) + ")"
    return result


def load_filter_rules(datalake_root: str) -> dict[str, str]:
    path = f"{datalake_root}/config/filter_rules.yaml"
    fs = s3fs.S3FileSystem()
    try:
        with fs.open(path, "r") as f:
            doc = yaml.safe_load(f) or {}
    except FileNotFoundError:
        # No rules file yet → no rows filtered. The bad_row_predicate macro
        # defaults each entity to FALSE (keep everything) on a missing key.
        return {}
    filters = doc.get("filters", {})
    return compile_filters(filters)

def _sync_asset(entity: str, import_table: str):
    # Shadow target while testing; cutover drops "_dagster" and swaps
    # rebuild_table -> reload_table into the live Import* tables (and confirms
    # which entities the product actually consumes).
    target = f"{import_table}_dagster"

    @dg.asset(
        name=entity,
        key_prefix=["mneye", "sync", "nextgen"],
        group_name="mneye_sync_nextgen",
        deps=[dg.AssetKey(["mneye", "bronze", "nextgen", "clean", entity])],
        metadata={"dagster/table_name": f"dbo.{target}"},
    )
    def _asset(
        context: AssetExecutionContext,
        mssql_target: MssqlTargetResource,
        customer_secrets: SecretsManagerSecretsResource,
    ) -> dg.MaterializeResult:
        cleaned = _cleaned_path(entity, run_date(context))
        # COLUMNS(...) drops dlt's internal _dlt_* columns from the load.
        source_sql = (
            f"SELECT COLUMNS(c -> NOT starts_with(c, '_dlt_')) "
            f"FROM read_parquet('{cleaned}', hive_partitioning=false)"
        )
        user, password = fetch_mssql_login(customer_secrets)
        conn = _connect_s3()
        try:
            rows = rebuild_table(
                conn,
                connection_string=mssql_target.duckdb_connection_string(user, password),
                source_sql=source_sql,
                target_table=target,
                log=context.log,
            )
        finally:
            conn.close()
        return dg.MaterializeResult(
            metadata={
                "dagster/table_name": dg.MetadataValue.text(f"dbo.{target}"),
                "dagster/row_count": dg.MetadataValue.int(rows),
            }
        )

    return _asset

@dg.multi_asset_sensor(
    monitored_assets=list(_SYNC_TARGET_KEYS),
    job=pipeline_complete_job,
    name="mneye_pipeline_complete_sensor",
    minimum_interval_seconds=30,
    default_status=dg.DefaultSensorStatus.RUNNING,
)
def mneye_pipeline_complete_sensor(
    context: dg.MultiAssetSensorEvaluationContext,
):
    records = context.latest_materialization_records_by_key()

    for key in _SYNC_TARGET_KEYS:
        if records.get(key) is None:
            return dg.SkipReason(f"waiting on {key.to_user_string()}")

    expected_batch: str | None = None
    for key in _SYNC_TARGET_KEYS:
        record = records[key]
        run = context.instance.get_run_by_id(record.run_id)
        batch = run.tags.get("date_segment") if run else None
        if batch is None:
            return dg.SkipReason(
                f"{key.to_user_string()} run {record.run_id} has no date_segment tag"
            )
        if expected_batch is None:
            expected_batch = batch
        elif batch != expected_batch:
            return dg.SkipReason(
                f"sync records span two date_segments: "
                f"{expected_batch} vs {batch}"
            )

    context.advance_all_cursors()
    return dg.RunRequest(tags={"date_segment": expected_batch})