mapper

Mapper module for converting SQLGlot parsed objects to CLOE metadata models.

AIDEV-NOTE: Maps SQLGlot expression objects (from CREATE TABLE/VIEW) to CLOE metadata models (Database, Schema, Table, Column). Attributes not directly mapped to CLOE model fields are stored in Table.options.

MappingError

Bases: Exception

Raised when required catalog or schema information is missing.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

class MappingError(Exception):
    """Raised when required catalog or schema information is missing."""

MappingResult

Container for mapping results.

Attributes:

Name	Type	Description
databases	dict[str | None, Database]	Nested CLOE model hierarchy keyed by database (catalog) name. Each value is a Database -> Schema -> Table -> Column tree. Extra SQL attributes are stored on each Table via Table.options. Keys may be None when ignore_missing=True was used.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

class MappingResult:
    """Container for mapping results.

    Attributes:
        databases: Nested CLOE model hierarchy keyed by database (catalog) name.
            Each value is a Database -> Schema -> Table -> Column tree.
            Extra SQL attributes are stored on each Table via Table.options.
            Keys may be ``None`` when ``ignore_missing=True`` was used.

    """

    def __init__(self) -> None:
        """Initialize empty mapping result."""
        self.databases: dict[str | None, Database] = {}

    def to_databases(self) -> Databases:
        """Return the mapped databases as a ``Databases`` repository instance.

        Runs the ``Databases`` uniqueness validator and gives callers access
        to lookup helpers like ``id_to_tables`` / ``get_table_and_schema``.
        """
        return Databases(databases=list(self.databases.values()))

__init__()

Initialize empty mapping result.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

def __init__(self) -> None:
    """Initialize empty mapping result."""
    self.databases: dict[str | None, Database] = {}

to_databases()

Return the mapped databases as a Databases repository instance.

Runs the Databases uniqueness validator and gives callers access to lookup helpers like id_to_tables / get_table_and_schema.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

def to_databases(self) -> Databases:
    """Return the mapped databases as a ``Databases`` repository instance.

    Runs the ``Databases`` uniqueness validator and gives callers access
    to lookup helpers like ``id_to_tables`` / ``get_table_and_schema``.
    """
    return Databases(databases=list(self.databases.values()))

export_sqlglot_schema_as_text(expr, output_path=None)

Export SQLGlot expression tree or parsed metadata as readable text.

Parameters:

Name	Type	Description	Default
expr	Expression | list[dict[str, Any]]	SQLGlot expression object or list of parsed metadata dicts.	required
output_path	str | None	Optional path to write the text file.	None

Returns:

Type	Description
str	String representation of the SQLGlot schema.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

def export_sqlglot_schema_as_text(expr: exp.Expression | list[dict[str, Any]], output_path: str | None = None) -> str:
    """Export SQLGlot expression tree or parsed metadata as readable text.

    Args:
        expr: SQLGlot expression object or list of parsed metadata dicts.
        output_path: Optional path to write the text file.

    Returns:
        String representation of the SQLGlot schema.

    """
    lines = []
    lines.append("=" * 80)
    lines.append("SQLGlot Expression Tree")
    lines.append("=" * 80)
    lines.append("")

    # Expression type
    lines.append(f"Expression Type: {type(expr).__name__}")
    lines.append("")

    # Handle both Expression objects and lists of metadata dicts
    lines.append("AST Structure:")
    lines.append("-" * 40)

    if isinstance(expr, list):
        # Format as list of metadata dicts
        lines.append(str(expr))
    else:
        # Pretty print the AST
        _format_expression_tree(expr, lines, indent=0)

    lines.append("")

    # SQL representation
    lines.append("SQL Representation:")
    lines.append("-" * 40)
    if isinstance(expr, list):
        lines.append("[List of metadata dictionaries]")
    else:
        try:
            lines.append(expr.sql(pretty=True))
        except Exception:
            lines.append(str(expr))

    text = "\n".join(lines)

    if output_path:
        Path(output_path).write_text(text, encoding="utf-8")

    return text

map_to_cloe_metadata(parsed_metadata, *, catalog=None, schema=None, ignore_missing=False)

Map parsed SQL metadata to CLOE metadata models.

Parameters:

Name	Type	Description	Default
parsed_metadata	list[dict[str, Any]]	List of metadata dictionaries from parse_sql().	required
catalog	str | None	Override catalog name. When provided and the SQL statement already contains a different catalog, the parameter value wins and a warning is logged.	None
schema	str | None	Override schema name. Same override/warning logic as catalog.	None
ignore_missing	bool	When True, missing catalog or schema values are silently accepted and the corresponding model objects are created with name=None. When False (default), a :class:MappingError is raised if catalog or schema cannot be determined.	False

Returns:

Type	Description
MappingResult	MappingResult with databases dict keyed by catalog name.
MappingResult	Extra SQL attributes are stored on each Table via Table.options.

Raises:

Type	Description
MappingError	If catalog or schema is missing and ignore_missing is False.

Source code in src/cloe_delta_table_manager/sql_parser/mapper.py

def map_to_cloe_metadata(
    parsed_metadata: list[dict[str, Any]],
    *,
    catalog: str | None = None,
    schema: str | None = None,
    ignore_missing: bool = False,
) -> MappingResult:
    """Map parsed SQL metadata to CLOE metadata models.

    Args:
        parsed_metadata: List of metadata dictionaries from parse_sql().
        catalog: Override catalog name.  When provided and the SQL
            statement already contains a different catalog, the
            parameter value wins and a warning is logged.
        schema: Override schema name.  Same override/warning logic
            as *catalog*.
        ignore_missing: When ``True``, missing catalog or schema
            values are silently accepted and the corresponding
            model objects are created with ``name=None``.
            When ``False`` (default), a :class:`MappingError` is
            raised if catalog or schema cannot be determined.

    Returns:
        MappingResult with databases dict keyed by catalog name.
        Extra SQL attributes are stored on each Table via Table.options.

    Raises:
        MappingError: If catalog or schema is missing and
            *ignore_missing* is ``False``.

    """
    hierarchy = _build_hierarchy(
        parsed_metadata,
        catalog=catalog,
        schema=schema,
        ignore_missing=ignore_missing,
    )
    dialect = parsed_metadata[0].get("dialect") if parsed_metadata else None

    result = MappingResult()
    for catalog_name, schemas_dict in hierarchy.items():
        result.databases[catalog_name] = _build_database(catalog_name, schemas_dict, dialect)
    return result