Text2SQL API

Models

maticlib.core.text2sql.models.ColumnSchema

Bases: BaseModel

maticlib.core.text2sql.models.TableSchema

Bases: BaseModel

to_ddl

to_ddl(dialect='sqlite')

Convert the TableSchema to a simplified DDL representation for prompts. We can use sqlglot or just basic formatting here.

Source code in maticlib/core/text2sql/models.py

def to_ddl(self, dialect: str = "sqlite") -> str:
    """
    Convert the TableSchema to a simplified DDL representation for prompts.
    We can use sqlglot or just basic formatting here.
    """
    lines = [f"CREATE TABLE {self.name} ("]
    for col in self.columns:
        pk = " PRIMARY KEY" if col.primary_key else ""
        fk = f" REFERENCES {col.foreign_key}" if col.foreign_key else ""
        lines.append(f"    {col.name} {col.data_type}{pk}{fk},")
    lines[-1] = lines[-1].rstrip(",")
    lines.append(");")
    return "\n".join(lines)

maticlib.core.text2sql.models.DatabaseSchema

Bases: BaseModel

to_prompt_string

to_prompt_string(subset_tables=None)

Returns the schema formatted for the LLM. subset_tables filters which tables are included (for schema pruning).

Source code in maticlib/core/text2sql/models.py

def to_prompt_string(self, subset_tables: Optional[List[str]] = None) -> str:
    """
    Returns the schema formatted for the LLM.
    subset_tables filters which tables are included (for schema pruning).
    """
    tables_to_include = self.tables
    if subset_tables:
        tables_to_include = [t for t in self.tables if t.name in subset_tables]

    return "\n\n".join([t.to_ddl() for t in tables_to_include])

Loaders

maticlib.core.text2sql.loaders.BaseSchemaLoader

Bases: ABC

Abstract base class for database schema loaders.

load_schema abstractmethod

load_schema(connection_string)

Loads and parses the schema from the given connection string.

Parameters:

Name	Type	Description	Default
connection_string	str	A database URI (e.g. sqlite:///my.db).	required

Returns:

Type	Description
DatabaseSchema	A fully populated DatabaseSchema object.

Source code in maticlib/core/text2sql/loaders.py

@abstractmethod
def load_schema(self, connection_string: str) -> DatabaseSchema:
    """
    Loads and parses the schema from the given connection string.

    Args:
        connection_string: A database URI (e.g. ``sqlite:///my.db``).

    Returns:
        A fully populated DatabaseSchema object.
    """
    pass

maticlib.core.text2sql.loaders.SQLAlchemySchemaLoader

Bases: BaseSchemaLoader

Reflects a live database schema using SQLAlchemy's inspect API.

load_schema

load_schema(connection_string)

Reflects all tables and columns from the connected database.

Parameters:

Name	Type	Description	Default
connection_string	str	A SQLAlchemy-compatible database URI.	required

Returns:

Type	Description
DatabaseSchema	A DatabaseSchema containing all tables and columns.

Raises:

Type	Description
MissingDependencyError	If sqlalchemy is not installed.
SchemaLoadError	If the schema cannot be reflected.

Source code in maticlib/core/text2sql/loaders.py

def load_schema(self, connection_string: str) -> DatabaseSchema:
    """
    Reflects all tables and columns from the connected database.

    Args:
        connection_string: A SQLAlchemy-compatible database URI.

    Returns:
        A DatabaseSchema containing all tables and columns.

    Raises:
        MissingDependencyError: If sqlalchemy is not installed.
        SchemaLoadError: If the schema cannot be reflected.
    """
    try:
        from sqlalchemy import create_engine, inspect
    except ImportError as e:
        raise MissingDependencyError(
            "sqlalchemy is required for SQLAlchemySchemaLoader. Install it with: pip install maticlib[text2sql]"
        ) from e

    try:
        engine = create_engine(connection_string)
        inspector = inspect(engine)

        tables = []
        for table_name in inspector.get_table_names():
            columns = []
            for col in inspector.get_columns(table_name):
                pk = col.get("primary_key", False)
                # Rough translation of SQLAlchemy type to string
                data_type = str(col["type"])

                columns.append(
                    ColumnSchema(
                        name=col["name"],
                        data_type=data_type,
                        primary_key=pk > 0 if isinstance(pk, int) else pk,
                    )
                )

            # Fetch foreign keys
            for fk in inspector.get_foreign_keys(table_name):
                constrained_columns = fk["constrained_columns"]
                referred_table = fk["referred_table"]
                referred_columns = fk["referred_columns"]

                if constrained_columns and referred_columns:
                    col_name = constrained_columns[0]
                    ref_str = f"{referred_table}.{referred_columns[0]}"

                    # Find column and add FK
                    for c in columns:
                        if c.name == col_name:
                            c.foreign_key = ref_str
                            break

            tables.append(TableSchema(name=table_name, columns=columns))

        return DatabaseSchema(tables=tables)
    except Exception as e:
        raise SchemaLoadError(
            f"Failed to load schema from {connection_string}: {e}"
        )

Executors

maticlib.core.text2sql.executors.BaseExecutor

Bases: ABC

Abstract base class for SQL query executors.

execute abstractmethod

execute(query)

Executes a SQL query and returns columns and rows.

Parameters:

Name	Type	Description	Default
query	str	A validated SELECT SQL query string.	required

Returns:

Type	Description
List[str]	A tuple of (columns, rows) where columns is a list of column name strings
List[tuple]	and rows is a list of row tuples.

Source code in maticlib/core/text2sql/executors.py

@abstractmethod
def execute(self, query: str) -> Tuple[List[str], List[tuple]]:
    """
    Executes a SQL query and returns columns and rows.

    Args:
        query: A validated SELECT SQL query string.

    Returns:
        A tuple of ``(columns, rows)`` where columns is a list of column name strings
        and rows is a list of row tuples.
    """
    pass

maticlib.core.text2sql.executors.SQLAlchemyExecutor

SQLAlchemyExecutor(connection_string, read_only=True)

Bases: BaseExecutor

Executes validated SQL queries against any SQLAlchemy-supported database.

Initializes the SQLAlchemyExecutor.

Parameters:

Name	Type	Description	Default
connection_string	str	A SQLAlchemy database URI.	required
read_only	bool	If True (default), opens SQLite connections in read-only mode.	True

Source code in maticlib/core/text2sql/executors.py

def __init__(self, connection_string: str, read_only: bool = True):
    """
    Initializes the SQLAlchemyExecutor.

    Args:
        connection_string: A SQLAlchemy database URI.
        read_only: If True (default), opens SQLite connections in read-only mode.
    """
    self.connection_string = connection_string
    self.read_only = read_only
    try:
        from sqlalchemy import create_engine

        # If using sqlite, we can use read-only uri
        if connection_string.startswith("sqlite:///") and read_only:
            if "?" in connection_string:
                uri = connection_string + "&mode=ro"
            else:
                uri = connection_string + "?mode=ro"
            self.engine = create_engine(
                uri,
                creator=lambda: __import__("sqlite3").connect(
                    uri.replace("sqlite:///", "file:"), uri=True
                ),
            )
        else:
            self.engine = create_engine(connection_string)
    except ImportError as e:
        raise MissingDependencyError(
            "sqlalchemy is required for SQLAlchemyExecutor. Install it with: pip install maticlib[text2sql]"
        ) from e

Tabular Ingestor

maticlib.core.text2sql.tabular_ingestor.TabularIngestor

TabularIngestor(
    connection_string="sqlite:///maticlib_data.db",
)

Source code in maticlib/core/text2sql/tabular_ingestor.py

def __init__(self, connection_string: str = "sqlite:///maticlib_data.db"):
    self.connection_string = connection_string
    try:
        from sqlalchemy import create_engine

        self.engine = create_engine(self.connection_string)
    except ImportError as e:
        raise MissingDependencyError(
            "sqlalchemy is required for TabularIngestor. Install it with: pip install maticlib[tabular]"
        ) from e

ingest_file

ingest_file(source, table_name=None, if_exists='replace')

Ingest a CSV, Excel, or Parquet file into the database. if_exists can be 'fail', 'replace', or 'append'.

Source code in maticlib/core/text2sql/tabular_ingestor.py

def ingest_file(
    self, source: str, table_name: Optional[str] = None, if_exists: str = "replace"
):
    """
    Ingest a CSV, Excel, or Parquet file into the database.
    if_exists can be 'fail', 'replace', or 'append'.
    """
    try:
        import pandas as pd
    except ImportError as e:
        raise MissingDependencyError(
            "pandas is required for TabularIngestor. Install it with: pip install maticlib[tabular]"
        ) from e

    if not os.path.exists(source):
        raise TabularIngestionError(f"File not found: {source}")

    if not table_name:
        # Default table name from filename
        base = os.path.basename(source)
        table_name = (
            os.path.splitext(base)[0].lower().replace(" ", "_").replace("-", "_")
        )

    ext = os.path.splitext(source)[1].lower()

    try:
        if ext == ".csv":
            df = pd.read_csv(source)
        elif ext in [".xls", ".xlsx"]:
            try:
                import openpyxl  # noqa
            except ImportError:
                raise MissingDependencyError(
                    "openpyxl is required for Excel files. pip install openpyxl"
                )
            df = pd.read_excel(source)
        elif ext == ".parquet":
            try:
                import pyarrow  # noqa
            except ImportError:
                raise MissingDependencyError(
                    "pyarrow is required for Parquet files. pip install pyarrow"
                )
            df = pd.read_parquet(source)
        else:
            raise TabularIngestionError(f"Unsupported file format: {ext}")

        df.to_sql(
            name=table_name, con=self.engine, if_exists=if_exists, index=False
        )
        return table_name
    except Exception as e:
        if isinstance(e, (MissingDependencyError, TabularIngestionError)):
            raise
        raise TabularIngestionError(f"Failed to ingest file {source}: {e}")

Guards

maticlib.core.text2sql.guards.SQLInjectionGuard

SQLInjectionGuard(allowed_dialect='sqlite')

Parses, validates, and transpiles SQL queries to prevent injection attacks.

Initializes the SQLInjectionGuard.

Parameters:

Name	Type	Description	Default
allowed_dialect	str	Target SQL dialect for transpilation (e.g. sqlite, postgres).	'sqlite'

Raises:

Type	Description
MissingDependencyError	If sqlglot is not installed.

Source code in maticlib/core/text2sql/guards.py

def __init__(self, allowed_dialect: str = "sqlite"):
    """
    Initializes the SQLInjectionGuard.

    Args:
        allowed_dialect: Target SQL dialect for transpilation (e.g. ``sqlite``, ``postgres``).

    Raises:
        MissingDependencyError: If sqlglot is not installed.
    """
    self.allowed_dialect = allowed_dialect
    try:
        import sqlglot
    except ImportError as e:
        raise MissingDependencyError(
            "sqlglot is required for SQLInjectionGuard. Install it with: pip install maticlib[text2sql]"
        ) from e

validate_and_format

validate_and_format(query)

Validates the SQL query against injection attacks and returns a cleanly formatted transpiled version of the query for the target dialect. Ensures the query is only performing SELECT operations.

Source code in maticlib/core/text2sql/guards.py

def validate_and_format(self, query: str) -> str:
    """
    Validates the SQL query against injection attacks and returns a cleanly formatted
    transpiled version of the query for the target dialect.
    Ensures the query is only performing SELECT operations.
    """
    import sqlglot
    from sqlglot.errors import ParseError

    try:
        # Parse the query. sqlglot will raise if syntax is invalid
        # We parse loosely, then transpile.
        expression = sqlglot.parse_one(query, read=None)

        # Check if it's a select query
        if not isinstance(expression, sqlglot.exp.Select):
            raise SQLInjectionError("Only SELECT queries are allowed.")

        # Additional checks could be added here (e.g., preventing access to system tables)

        # Transpile and format
        safe_query = expression.sql(dialect=self.allowed_dialect, pretty=True)
        return safe_query

    except ParseError as e:
        raise SQLValidationError(f"Invalid SQL syntax: {e}")
    except SQLInjectionError:
        raise
    except Exception as e:
        raise SQLValidationError(f"Failed to validate SQL: {e}")