cdm_data_loaders.converters.emitters.dlt ¶

class DltEmitter(BaseModel):
    """Configure typed emission with JSON-to-dlt compatibility defaults."""

    model_config = ConfigDict(frozen=True)

    schema_name: str
    skip_nested_types: bool = False
    max_nesting: PositiveInt = Field(default=10)
    write_disposition: str | None = None
    flatten_scalars: bool = True

    @field_validator("schema_name")
    @classmethod
    def _validate_schema_name(cls, value: str) -> str:
        """Reject blank schema names."""
        return validate_non_empty_schema_name(value)

    def emit(self, document: SchemaDocument) -> TStoredSchema:
        """Emit one document as a stored schema with one root table."""
        root = document.root
        if not _is_json(root) and root.type != "object":
            msg = "A dlt table requires an object-compatible root"
            raise ConversionError(msg)
        native = root.provenance is not None and root.provenance.source == "dlt"
        name = document.name if native and document.name is not None else self.schema_name
        tree = self._table(root, name, None, 0, native=native)
        stored: TStoredSchema = {
            "name": self.schema_name,
            "version": 1,
            "previous_hashes": [],
            "engine_version": SCHEMA_ENGINE_VERSION,
            "tables": flatten_nodes({tree.name: tree}),
        }
        if native and document.provenance is not None and document.provenance.source == "dlt":
            envelope = document.provenance.metadata.get("envelope")
            if isinstance(envelope, Mapping):
                stored.update(
                    {key: mutable_value(value) for key, value in envelope.items() if key not in {"name", "tables"}}
                )
        return stored

    def to_schema(self, document: SchemaDocument) -> Schema:
        """Build a live schema, installing every parent before its children."""
        stored = self.emit(document)
        schema = Schema(stored["name"])
        for name, table in stored["tables"].items():
            table["name"] = name
            for column_name, column in table.setdefault("columns", {}).items():
                column["name"] = column_name
        for table in tables_parents_first(stored["tables"]):
            schema.update_table(table, normalize_identifiers=False)
        return schema

    def to_yaml(self, document: SchemaDocument) -> str:
        """Serialize the emitted stored schema without sorting table keys."""
        return yaml.safe_dump(self.emit(document), allow_unicode=True, default_flow_style=False, sort_keys=False)

    def _table(
        self, node: TypedNode, name: str, parent: str | None, depth: int, *, native: bool = False
    ) -> DltTableNode:
        """Recursively derive columns and child tables from object fields."""
        columns: dict[str, TColumnSchema] = {}
        children: dict[str, DltTableNode] = {}
        for prop in node.properties:
            child = self._child(prop, name, depth, native=native)
            if child is None:
                columns[prop.name] = self._column(prop, native=native)
            else:
                children[child.key] = child
        table = new_table(
            table_name=name,
            parent_table_name=parent,
            write_disposition=self.write_disposition if parent is None else None,
        )
        table["columns"] = columns
        if "description" in node.annotations:
            table["description"] = cast("str", mutable_value(node.annotations["description"]))
        tree = DltTableNode(name, parent, child_key(parent, name) if parent else name, table, children)
        if native:
            tree = _restore_table(tree, node)
        if parent is None and self.write_disposition is not None:
            tree.table["write_disposition"] = self.write_disposition
        return tree

    def _child(self, prop: SchemaField, parent: str, depth: int, *, native: bool = False) -> DltTableNode | None:
        """Flatten structures while retaining the legacy object-only depth limit."""
        node = prop.node
        if self.skip_nested_types:
            return None
        kind = _resolve_type(node)
        name = child_table_name(parent, prop.name)
        metadata = _dlt_table_metadata(node) if native else None
        if metadata is not None and node.provenance is not None:
            original = node.provenance.path[1]
            if (
                isinstance(original, str)
                and metadata.get("parent") == parent
                and child_key(parent, original) == prop.name
            ):
                name = original
        child = None
        if kind == "object":
            if depth < self.max_nesting:
                child = self._table(node, name, parent, depth + 1, native=native)
            else:
                logger.warning(
                    "Schema nesting limit (%d) reached at %r; ignoring object child.", self.max_nesting, prop.name
                )
        elif kind == "array" and (item := _single_item(node)) is not None:
            if _resolve_type(item) == "object":
                child = self._table(item, name, parent, depth + 1, native=native)
            elif self.flatten_scalars:
                value = (
                    new_column("value", _scalar_type(item))
                    if _is_json(node)
                    else self._column(SchemaField("value", item), native=native)
                )
                table = new_table(name, parent_table_name=parent)
                table["columns"] = {"value": value}
                child = DltTableNode(name, parent, child_key(parent, name), table)
        if child is not None:
            if native:
                child = _restore_table(child, node)
            if not _is_json(node) and "description" in prop.annotations:
                child.table["description"] = cast("str", mutable_value(prop.annotations["description"]))
        return child

    def _column(self, prop: SchemaField, *, native: bool = False) -> TColumnSchema:
        """Derive column types and JSON presence-based nullability."""
        node = prop.node
        if _boolean_schema(node):
            logger.warning("Boolean schema for property %r carries no type; using data_type 'text'.", prop.name)
        nullable = not prop.required if _is_json(node) or node.nullable is None else node.nullable
        column = new_column(prop.name, data_type=_scalar_type(node), nullable=nullable)
        if not _is_json(node):
            column.update(_column_hints(node))
        annotations = node.annotations if _is_json(node) else {**node.annotations, **prop.annotations}
        if "description" in annotations:
            column["description"] = cast("str", mutable_value(annotations["description"]))
        return _restore_column(column, node) if native else column