process

High-level wrapper that combines SQL parsing and CLOE metadata mapping.

Provides a single entry-point that accepts a file path, a directory path, or a raw SQL string, and returns a :class:MappingResult in one call.

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)