types.SQLAlchemySource

SQLAlchemySource(engine, table_name)

A DataSource implementation that supports multiple SQL databases via SQLAlchemy.

Supports various databases including PostgreSQL, MySQL, SQLite, Snowflake, and Databricks.

Attributes

Name	Description
engine	The SQLAlchemy engine for this data source.

Name	Description
cleanup	No-op: the SQLAlchemy engine is owned by the caller.
execute_query	Execute SQL query and return results as DataFrame.
get_data	Return the unfiltered data as a DataFrame.
get_db_type	Get the database type.
get_schema	Generate schema information from database table.
get_semantic_views_description	Get information about semantic views (if any) for the system prompt.
test_query	Test query by fetching only one row.

types.SQLAlchemySource.cleanup()

No-op: the SQLAlchemy engine is owned by the caller.

Dispose it yourself with engine.dispose() when your application shuts down.

types.SQLAlchemySource.execute_query(query)

Execute SQL query and return results as DataFrame.

Name	Type	Description	Default
query	str	SQL query to execute	required

Name	Type	Description
	`nw`.`DataFrame`	Query results as narwhals DataFrame

Name	Type	Description
	`UnsafeQueryError`	If the query starts with a disallowed SQL operation

types.SQLAlchemySource.get_data()

Return the unfiltered data as a DataFrame.

Name	Type	Description
	`nw`.`DataFrame`	The complete dataset as narwhals DataFrame

types.SQLAlchemySource.get_db_type()

Get the database type.

Returns the specific database type (e.g., POSTGRESQL, MYSQL, SQLITE) by inspecting the SQLAlchemy engine. Removes ” SQL” suffix if present.

types.SQLAlchemySource.get_schema(categorical_threshold)

Generate schema information from database table.

Name	Type	Description	Default
categorical_threshold	int	Maximum number of unique values for a text column to be considered categorical	required

Name	Type	Description
	str	String describing the schema

types.SQLAlchemySource.get_semantic_views_description()

Get information about semantic views (if any) for the system prompt.

types.SQLAlchemySource.test_query(query, *, require_all_columns=False)

Test query by fetching only one row.

Name	Type	Description	Default
query	str	SQL query to test	required
require_all_columns	bool	If True, validates that result includes all original table columns	`False`

Name	Type	Description
	`nw`.`DataFrame`	Query results with at most one row

Name	Type	Description
	`UnsafeQueryError`	If the query starts with a disallowed SQL operation
	`MissingColumnsError`	If require_all_columns is True and result is missing required columns