Index

SQL Parser module for CLOE Delta Table Manager.

Public API for SQL parsing. Uses sqlglot to extract metadata from CREATE TABLE and CREATE VIEW statements and map them to CLOE metadata models.

Example usage::

from cloe_delta_table_manager.sql_parser import process_sql

# From a single file
result = process_sql("tables/users.sql")

# From a directory
result = process_sql("sql/", recursive=True)

# From a raw SQL string
result = process_sql(sql="CREATE TABLE cat.sch.t (id INT)")

for name, db in result.databases.items():
    for schema in db.schemas:
        for table in schema.tables:
            print(f"{name}.{schema.name}.{table.name}")

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()))

ParseError

Bases: Exception

Raise when SQL cannot be parsed.

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

class ParseError(Exception):
    """Raise when SQL cannot be parsed."""

    def __init__(self, message: str, sql: str | None = None) -> None:
        """Initialize parse error.

        Args:
            message: Error message.
            sql: Optional SQL that caused the error.

        """
        self.sql = sql
        super().__init__(message)

__init__(message, sql=None)

Initialize parse error.

Parameters:

Name	Type	Description	Default
message	str	Error message.	required
sql	str | None	Optional SQL that caused the error.	None

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

def __init__(self, message: str, sql: str | None = None) -> None:
    """Initialize parse error.

    Args:
        message: Error message.
        sql: Optional SQL that caused the error.

    """
    self.sql = sql
    super().__init__(message)

UnsupportedStatementError

Bases: ParseError

Raise when a SQL file contains a statement the tool does not yet support.

AIDEV-NOTE: Subclasses ParseError so the CLI's existing except ParseError handlers surface it without changes. Only the solution SQL folder enforces this — migration scripts may contain arbitrary SQL.

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

class UnsupportedStatementError(ParseError):
    """Raise when a SQL file contains a statement the tool does not yet support.

    AIDEV-NOTE: Subclasses ParseError so the CLI's existing ``except ParseError``
    handlers surface it without changes. Only the solution SQL folder enforces
    this — migration scripts may contain arbitrary SQL.
    """

check_reference_qualifications(file_path, dialect='databricks')

Check that every table reference in a SQL file is fully qualified.

Where check_statement_qualifications inspects only CREATE TABLE/VIEW targets, this inspects every table reference in any statement — INSERT, SELECT, DELETE, MERGE, ALTER, DROP, … — so migration scripts can be held to the same database.schema.name rule as the solution.

Common-table-expression (WITH) names are treated as local and skipped, as are USE statements (their target is a namespace, not a table reference).

Parameters:

Name	Type	Description	Default
file_path	str	Path to SQL file.	required
dialect	str	SQL dialect to use for parsing.	'databricks'

Returns:

Type	Description
list[str]	One message per distinct under-qualified reference; empty when every
list[str]	reference carries both a database (catalog) and a schema.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
ParseError	If SQL cannot be parsed.

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

def check_reference_qualifications(file_path: str, dialect: str = "databricks") -> list[str]:
    """Check that every table reference in a SQL file is fully qualified.

    Where ``check_statement_qualifications`` inspects only CREATE TABLE/VIEW
    targets, this inspects *every* table reference in *any* statement — INSERT,
    SELECT, DELETE, MERGE, ALTER, DROP, … — so migration scripts can be held to
    the same ``database.schema.name`` rule as the solution.

    Common-table-expression (``WITH``) names are treated as local and skipped, as
    are ``USE`` statements (their target is a namespace, not a table reference).

    Args:
        file_path: Path to SQL file.
        dialect: SQL dialect to use for parsing.

    Returns:
        One message per distinct under-qualified reference; empty when every
        reference carries both a database (catalog) and a schema.

    Raises:
        FileNotFoundError: If file does not exist.
        ParseError: If SQL cannot be parsed.

    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")
    sql = path.read_text(encoding="utf-8")
    try:
        statements = sqlglot.parse(sql, dialect=dialect)
    except sqlglot.errors.ParseError as e:
        raise ParseError(f"Failed to parse SQL: {e}", sql=sql) from e

    errors: list[str] = []
    seen: set[str] = set()
    for statement in statements:
        if statement is None or isinstance(statement, exp.Use):
            continue
        local_names = {cte.alias_or_name for cte in statement.find_all(exp.CTE)}
        for table in statement.find_all(exp.Table):
            if not table.name or table.name in local_names:
                continue
            missing = []
            if not table.catalog:
                missing.append("database")
            if not table.db:
                missing.append("schema")
            if not missing:
                continue
            ref = ".".join(part for part in (table.catalog, table.db, table.name) if part)
            message = f"'{ref}': reference missing {' and '.join(missing)}"
            if message not in seen:
                seen.add(message)
                errors.append(message)
    return errors

check_statement_qualifications(file_path, dialect='databricks')

Check that all CREATE TABLE/VIEW statements have both schema and database set.

Parameters:

Name	Type	Description	Default
file_path	str	Path to SQL file.	required
dialect	str	SQL dialect to use for parsing.	'databricks'

Returns:

Type	Description
list[str]	List of error messages for statements missing database (catalog) or schema.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
ParseError	If SQL cannot be parsed.

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

def check_statement_qualifications(file_path: str, dialect: str = "databricks") -> list[str]:
    """Check that all CREATE TABLE/VIEW statements have both schema and database set.

    Args:
        file_path: Path to SQL file.
        dialect: SQL dialect to use for parsing.

    Returns:
        List of error messages for statements missing database (catalog) or schema.

    Raises:
        FileNotFoundError: If file does not exist.
        ParseError: If SQL cannot be parsed.

    """
    metadata_list = parse_sql_file(file_path, dialect=dialect)
    errors: list[str] = []
    for metadata in metadata_list:
        name = metadata.get("fully_qualified_name") or metadata.get("name") or "unknown"
        missing = []
        if not metadata.get("catalog"):
            missing.append("database")
        if not metadata.get("schema"):
            missing.append("schema")
        if missing:
            errors.append(f"'{name}': missing {' and '.join(missing)}")
    return errors

check_unique_object_identifiers(directory_path, dialect='databricks', recursive=True, pattern='*.sql')

Check that each object identifier is unique across a directory of SQL files.

The identity of an object is the combination of its object type, database (catalog), schema, and name. Two CREATE statements sharing all four — whether in the same file or different files — are reported as a collision.

Parameters:

Name	Type	Description	Default
directory_path	str	Path to the directory to scan.	required
dialect	str	SQL dialect to use for parsing.	'databricks'
recursive	bool	If True, scan subdirectories recursively.	True
pattern	str	Glob pattern for SQL files.	'*.sql'

Returns:

Type	Description
list[str]	One error message per duplicated identifier, naming the files involved.
list[str]	Empty when every object identifier is unique. File paths are relative to
list[str]	directory_path.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.

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

def check_unique_object_identifiers(
    directory_path: str,
    dialect: str = "databricks",
    recursive: bool = True,
    pattern: str = "*.sql",
) -> list[str]:
    """Check that each object identifier is unique across a directory of SQL files.

    The identity of an object is the combination of its object type, database
    (catalog), schema, and name. Two CREATE statements sharing all four — whether
    in the same file or different files — are reported as a collision.

    Args:
        directory_path: Path to the directory to scan.
        dialect: SQL dialect to use for parsing.
        recursive: If True, scan subdirectories recursively.
        pattern: Glob pattern for SQL files.

    Returns:
        One error message per duplicated identifier, naming the files involved.
        Empty when every object identifier is unique. File paths are relative to
        *directory_path*.

    Raises:
        FileNotFoundError: If the directory does not exist.

    """
    root = Path(directory_path)
    parsed = parse_sql_directory(directory_path, dialect=dialect, recursive=recursive, pattern=pattern)

    occurrences: dict[tuple[Any, Any, Any, Any], list[str]] = {}
    for file_path, metadata_list in parsed.items():
        try:
            location = str(Path(file_path).relative_to(root))
        except ValueError:
            location = file_path
        for metadata in metadata_list:
            key = (
                metadata.get("object_type"),
                metadata.get("catalog"),
                metadata.get("schema"),
                metadata.get("name"),
            )
            occurrences.setdefault(key, []).append(location)

    errors: list[str] = []
    for key in sorted(occurrences, key=lambda k: tuple(part or "" for part in k)):
        locations = occurrences[key]
        if len(locations) <= 1:
            continue
        object_type, catalog, schema, name = key
        fqn = ".".join(part for part in (catalog, schema, name) if part) or name or "unknown"
        files = ", ".join(sorted(set(locations)))
        errors.append(f"duplicate {object_type or 'object'} '{fqn}' defined {len(locations)} times in: {files}")
    return errors

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

get_mapped_attributes()

Get all successfully mapped attributes.

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

def get_mapped_attributes() -> dict[str, Any]:
    """Get all successfully mapped attributes."""
    mapped: dict[str, Any] = {}

    for category, mappings in SQLGLOT_TO_CLOE_MAPPING.items():
        if isinstance(mappings, dict) and "mapped" in mappings:
            mapped[category] = mappings["mapped"]

    return mapped

get_mapping_documentation()

Get the complete mapping documentation.

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

def get_mapping_documentation() -> dict[str, Any]:
    """Get the complete mapping documentation."""
    return SQLGLOT_TO_CLOE_MAPPING

get_undefined_mappings()

Get all undefined/unmapped attributes.

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

def get_undefined_mappings() -> dict[str, Any]:
    """Get all undefined/unmapped attributes."""
    undefined: dict[str, Any] = {}

    for category, mappings in SQLGLOT_TO_CLOE_MAPPING.items():
        if isinstance(mappings, dict) and "undefined" in mappings:
            undefined[category] = mappings["undefined"]

    return undefined

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

parse_sql(sql, dialect='databricks')

Parse SQL string and extract metadata for all supported statements.

Parameters:

Name	Type	Description	Default
sql	str	SQL string containing one or more statements.	required
dialect	str	SQL dialect to use for parsing (default: "databricks").	'databricks'

Returns:

Type	Description
list[dict[str, Any]]	List of metadata dictionaries, one per supported statement.

Raises:

Type	Description
ParseError	If SQL cannot be parsed.
UnsupportedStatementError	If the SQL contains a statement other than CREATE TABLE/VIEW. Unsupported statements are never silently dropped.

Example

metadata = parse_sql("CREATE TABLE t (id INT)", dialect="databricks") print(metadata[0]["object_type"]) 'TABLE'

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

def parse_sql(sql: str, dialect: str = "databricks") -> list[dict[str, Any]]:
    """Parse SQL string and extract metadata for all supported statements.

    Args:
        sql: SQL string containing one or more statements.
        dialect: SQL dialect to use for parsing (default: "databricks").

    Returns:
        List of metadata dictionaries, one per supported statement.

    Raises:
        ParseError: If SQL cannot be parsed.
        UnsupportedStatementError: If the SQL contains a statement other than
            CREATE TABLE/VIEW. Unsupported statements are never silently dropped.

    Example:
        >>> metadata = parse_sql("CREATE TABLE t (id INT)", dialect="databricks")
        >>> print(metadata[0]["object_type"])
        'TABLE'

    """
    try:
        expressions = sqlglot.parse(sql, dialect=dialect)
    except sqlglot.errors.ParseError as e:
        raise ParseError(f"Failed to parse SQL: {e}", sql=sql) from e

    results = []
    for expr in expressions:
        # None / Semicolon are no-op nodes (empty statements, trailing comments).
        if expr is None or isinstance(expr, exp.Semicolon):
            continue
        # Extract supported objects; fail loudly on anything else rather than
        # silently dropping it. The CLI validation gate usually reports this
        # first; compare and programmatic callers rely on this backstop.
        if isinstance(expr, exp.Create) and str(expr.args.get("kind", "")).upper() in SUPPORTED_STATEMENT_KINDS:
            results.append(_extract_create_metadata(expr, dialect))
        else:
            raise UnsupportedStatementError(
                f"unsupported statement: {_describe_statement(expr)}. Only CREATE TABLE and CREATE VIEW are supported.",
                sql=sql,
            )

    return results

parse_sql_directory(directory_path, dialect='databricks', recursive=False, pattern='*.sql')

Parse all SQL files in a directory and extract metadata.

Parameters:

Name	Type	Description	Default
directory_path	str	Path to directory containing SQL files.	required
dialect	str	SQL dialect to use for parsing (default: "databricks").	'databricks'
recursive	bool	If True, search subdirectories recursively.	False
pattern	str	Glob pattern for SQL files (default: "*.sql").	'*.sql'

Returns:

Name	Type	Description
	dict[str, list[dict[str, Any]]]	Dictionary mapping file paths to their parsed metadata.
Format	dict[str, list[dict[str, Any]]]	{file_path: [metadata_dict, ...]}

Raises:

Type	Description
FileNotFoundError	If directory does not exist.
NotADirectoryError	If the path is not a directory.
ParseError	If any file has invalid SQL syntax.
UnsupportedStatementError	If any file contains a statement other than CREATE TABLE/VIEW. Files are never silently skipped — a malformed or unsupported file must be removed from the directory first.

Example

results = parse_sql_directory("./sql_files", recursive=True) for file_path, metadata_list in results.items(): ... print(f"{file_path}: {len(metadata_list)} objects")

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

def parse_sql_directory(
    directory_path: str,
    dialect: str = "databricks",
    recursive: bool = False,
    pattern: str = "*.sql",
) -> dict[str, list[dict[str, Any]]]:
    """Parse all SQL files in a directory and extract metadata.

    Args:
        directory_path: Path to directory containing SQL files.
        dialect: SQL dialect to use for parsing (default: "databricks").
        recursive: If True, search subdirectories recursively.
        pattern: Glob pattern for SQL files (default: "*.sql").

    Returns:
        Dictionary mapping file paths to their parsed metadata.
        Format: {file_path: [metadata_dict, ...]}

    Raises:
        FileNotFoundError: If directory does not exist.
        NotADirectoryError: If the path is not a directory.
        ParseError: If any file has invalid SQL syntax.
        UnsupportedStatementError: If any file contains a statement other than
            CREATE TABLE/VIEW. Files are never silently skipped — a malformed or
            unsupported file must be removed from the directory first.

    Example:
        >>> results = parse_sql_directory("./sql_files", recursive=True)
        >>> for file_path, metadata_list in results.items():
        ...     print(f"{file_path}: {len(metadata_list)} objects")

    """
    path = Path(directory_path)
    if not path.exists():
        raise FileNotFoundError(f"Directory not found: {directory_path}")
    if not path.is_dir():
        raise NotADirectoryError(f"Path is not a directory: {directory_path}")

    results: dict[str, list[dict[str, Any]]] = {}
    glob_method = path.rglob if recursive else path.glob

    for sql_file in sorted(glob_method(pattern)):
        if sql_file.is_file():
            metadata = parse_sql_file(str(sql_file), dialect=dialect)
            if metadata:  # Files with no objects (e.g. comments only) contribute nothing.
                results[str(sql_file)] = metadata

    return results

parse_sql_file(file_path, dialect='databricks')

Parse SQL file and extract metadata.

Parameters:

Name	Type	Description	Default
file_path	str	Path to SQL file.	required
dialect	str	SQL dialect to use for parsing.	'databricks'

Returns:

Type	Description
list[dict[str, Any]]	List of metadata dictionaries.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
ParseError	If SQL cannot be parsed.

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

def parse_sql_file(file_path: str, dialect: str = "databricks") -> list[dict[str, Any]]:
    """Parse SQL file and extract metadata.

    Args:
        file_path: Path to SQL file.
        dialect: SQL dialect to use for parsing.

    Returns:
        List of metadata dictionaries.

    Raises:
        FileNotFoundError: If file does not exist.
        ParseError: If SQL cannot be parsed.

    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")
    sql = path.read_text(encoding="utf-8")
    return parse_sql(sql, dialect=dialect)

print_mapping_summary()

Generate a human-readable mapping summary.

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

def print_mapping_summary() -> str:
    """Generate a human-readable mapping summary."""
    lines = _format_header()
    lines.extend(_format_object_mappings(SQLGLOT_TO_CLOE_MAPPING["object_mappings"]))

    for category in ["table", "column", "schema", "database"]:
        key = f"{category}_parameter_mappings"
        if key in SQLGLOT_TO_CLOE_MAPPING:
            lines.extend(_format_parameter_section(category, SQLGLOT_TO_CLOE_MAPPING[key]))

    return "\n".join(lines)

process_sql(source_path=None, *, sql=None, dialect='databricks', recursive=False, pattern='*.sql', catalog=None, schema=None, ignore_missing=False)

Parse SQL and map to CLOE metadata in one step.

Provide either source_path (a file or directory) or sql (a raw SQL string). When both or neither are given a ValueError is raised.

• If source_path points to a file, it is read and parsed.
• If source_path points to a directory, every matching SQL file inside it is parsed and the results are merged into one :class:MappingResult. Use recursive and pattern to control which files are included.
• If sql is given (source_path must be None), the string is parsed directly.

Parameters:

Name	Type	Description	Default
source_path	str | Path | None	Path to a .sql file or a directory of SQL files.	None
sql	str | None	Raw SQL string containing CREATE TABLE / VIEW statements.	None
dialect	str	SQL dialect for parsing (default: "databricks").	'databricks'
recursive	bool	Search subdirectories when source_path is a directory.	False
pattern	str	Glob pattern for SQL files (default: "*.sql").	'*.sql'
catalog	str | None	Override catalog name for all statements.	None
schema	str | None	Override schema name for all statements.	None
ignore_missing	bool	Accept missing catalog/schema (created with name=None instead of raising).	False

Returns:

Type	Description
MappingResult	MappingResult with databases dict keyed by catalog name.

Raises:

Type	Description
ValueError	If both or neither of source_path / sql are given.
FileNotFoundError	If source_path does not exist.
NotADirectoryError	If source_path exists but is neither file nor directory.
ParseError	If the SQL cannot be parsed.
MappingError	If catalog/schema is missing and ignore_missing is False.

Examples:

>>> result = process_sql(sql="CREATE TABLE cat.sch.t (id INT)")
>>> result.databases["cat"].schemas[0].tables[0].name
't'

>>> result = process_sql("tables/users.sql")

>>> result = process_sql("sql/", recursive=True)

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

def process_sql(  # noqa: PLR0913
    source_path: str | Path | None = None,
    *,
    sql: str | None = None,
    dialect: str = "databricks",
    recursive: bool = False,
    pattern: str = "*.sql",
    catalog: str | None = None,
    schema: str | None = None,
    ignore_missing: bool = False,
) -> MappingResult:
    """Parse SQL and map to CLOE metadata in one step.

    Provide **either** ``source_path`` (a file or directory) **or** ``sql``
    (a raw SQL string).  When both or neither are given a ``ValueError``
    is raised.

    * If ``source_path`` points to a **file**, it is read and parsed.
    * If ``source_path`` points to a **directory**, every matching SQL file
      inside it is parsed and the results are merged into one
      :class:`MappingResult`.  Use *recursive* and *pattern* to control
      which files are included.
    * If ``sql`` is given (``source_path`` must be ``None``), the string
      is parsed directly.

    Args:
        source_path: Path to a ``.sql`` file or a directory of SQL files.
        sql: Raw SQL string containing CREATE TABLE / VIEW statements.
        dialect: SQL dialect for parsing (default: ``"databricks"``).
        recursive: Search subdirectories when *source_path* is a directory.
        pattern: Glob pattern for SQL files (default: ``"*.sql"``).
        catalog: Override catalog name for all statements.
        schema: Override schema name for all statements.
        ignore_missing: Accept missing catalog/schema (created with
            ``name=None`` instead of raising).

    Returns:
        MappingResult with databases dict keyed by catalog name.

    Raises:
        ValueError: If both or neither of *source_path* / *sql* are given.
        FileNotFoundError: If *source_path* does not exist.
        NotADirectoryError: If *source_path* exists but is neither file
            nor directory.
        ParseError: If the SQL cannot be parsed.
        MappingError: If catalog/schema is missing and *ignore_missing*
            is ``False``.

    Examples:
        >>> result = process_sql(sql="CREATE TABLE cat.sch.t (id INT)")
        >>> result.databases["cat"].schemas[0].tables[0].name
        't'

        >>> result = process_sql("tables/users.sql")

        >>> result = process_sql("sql/", recursive=True)

    """
    if (source_path is None) == (sql is None):
        msg = "Provide exactly one of 'source_path' or 'sql', not both."
        raise ValueError(msg)

    def _map(parsed: list[dict[str, Any]]) -> MappingResult:
        return map_to_cloe_metadata(parsed, catalog=catalog, schema=schema, ignore_missing=ignore_missing)

    # --- raw SQL string ---------------------------------------------------
    if sql is not None:
        return _map(parse_sql(sql, dialect=dialect))

    # --- file or directory ------------------------------------------------
    assert source_path is not None  # mypy guard
    path = Path(source_path)

    if not path.exists():
        raise FileNotFoundError(f"Source path not found: {source_path}")

    if path.is_file():
        return _map(parse_sql_file(str(path), dialect=dialect))

    if path.is_dir():
        file_results = parse_sql_directory(
            str(path),
            dialect=dialect,
            recursive=recursive,
            pattern=pattern,
        )
        all_metadata: list[dict[str, Any]] = []
        for metadata_list in file_results.values():
            all_metadata.extend(metadata_list)
        return _map(all_metadata)

    msg = f"Source path is neither a file nor a directory: {source_path}"
    raise NotADirectoryError(msg)

validate_sql_file(file_path, dialect='databricks', *, enforce_supported=False)

Validate SQL syntax in a file using strict parsing.

Unlike parse_sql_file, this validates ALL statement types (CREATE, ALTER, etc.), not just DDL metadata extraction.

Parameters:

Name	Type	Description	Default
file_path	str	Path to SQL file.	required
dialect	str	SQL dialect to use for parsing.	'databricks'
enforce_supported	bool	When True, reject any statement that is not a CREATE TABLE / CREATE VIEW. Used for the solution SQL folder, which may only define supported objects; migration scripts leave this False so they can contain arbitrary SQL.	False

Returns:

Type	Description
list[str]	List of statement descriptions (e.g. "Create", "Alter") for valid statements.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
ParseError	If any SQL statement has invalid syntax.
UnsupportedStatementError	If enforce_supported is True and the file contains a statement other than CREATE TABLE/VIEW.

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

def validate_sql_file(file_path: str, dialect: str = "databricks", *, enforce_supported: bool = False) -> list[str]:
    """Validate SQL syntax in a file using strict parsing.

    Unlike parse_sql_file, this validates ALL statement types (CREATE, ALTER, etc.),
    not just DDL metadata extraction.

    Args:
        file_path: Path to SQL file.
        dialect: SQL dialect to use for parsing.
        enforce_supported: When True, reject any statement that is not a
            ``CREATE TABLE`` / ``CREATE VIEW``. Used for the solution SQL folder,
            which may only define supported objects; migration scripts leave this
            ``False`` so they can contain arbitrary SQL.

    Returns:
        List of statement descriptions (e.g. "Create", "Alter") for valid statements.

    Raises:
        FileNotFoundError: If file does not exist.
        ParseError: If any SQL statement has invalid syntax.
        UnsupportedStatementError: If ``enforce_supported`` is True and the file
            contains a statement other than CREATE TABLE/VIEW.

    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")
    sql = path.read_text(encoding="utf-8")
    try:
        expressions = sqlglot.parse(sql, dialect=dialect, error_level=sqlglot.ErrorLevel.RAISE)
    except sqlglot.errors.ParseError as e:
        raise ParseError(f"Syntax error in {file_path}: {e}", sql=sql) from e

    if enforce_supported:
        _check_supported_statements(expressions, sql)

    return [type(expr).__name__ for expr in expressions if expr is not None]