parser

Core SQL parsing functionality using sqlglot.

AIDEV-NOTE: Main parser module. Uses sqlglot.parse() exclusively. Extracts metadata as dictionaries which are then mapped to CLOE metadata models.

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

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)

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]