unitxt.db_utils module¶

class unitxt.db_utils.DatabaseConnector(db_config: SQLDatabase)[source]¶

Bases: ABC

Abstract base class for database connectors.

abstract execute_query(query: str) → Any[source]¶: Abstract method to execute a query against the database.

abstract get_table_schema() → str[source]¶: Abstract method to get database schema.

class unitxt.db_utils.InMemoryDatabaseConnector(db_config: SQLDatabase)[source]¶

Database connector for mocking databases with in-memory data structures.

execute_query(query: str) → Any[source]¶: Simulates executing a query against the mock database.

get_table_schema(select_tables: List[str] | None = None) → str[source]¶: Generates a mock schema from the tables structure.

class unitxt.db_utils.LocalSQLiteConnector(db_config: SQLDatabase)[source]¶

Database connector for SQLite databases.

download_database(db_id)[source]¶: Downloads the database from huggingface if needed.

execute_query(query: str) → Any[source]¶: Executes a query against the SQLite database.

get_db_file_path(db_id)[source]¶: Gets the local path of a downloaded database file.

class unitxt.db_utils.RemoteDatabaseConnector(db_config: SQLDatabase)[source]¶

Database connector for remote databases accessed via HTTP.

execute_query(query: str) → Any[source]¶: Executes a query against the remote database, with retries for certain exceptions.

unitxt.db_utils.execute_query_local(db_path: str, query: str) → Any¶: Executes a query against the SQLite database.

unitxt.db_utils.execute_query_remote(api_url: str, database_id: str, api_key: str, query: str, retryable_exceptions: tuple = (<class 'requests.exceptions.ConnectionError'>, <class 'requests.exceptions.ReadTimeout'>), max_retries: int = 3, retry_delay: int = 5, timeout: int = 30) → dict | None¶: Executes a query against the remote database, with retries for certain exceptions.

unitxt.db_utils.get_db_connector(db_type: str)[source]¶: Creates and returns the appropriate DatabaseConnector instance based on db_type.