database
core.database
Database connection and query management.
Classes
| Name | Description |
|---|---|
| DatabaseConnection | Database connection manager. |
DatabaseConnection
core.database.DatabaseConnection(self, settings, query_timer=None, console=None)Database connection manager.
Methods
| Name | Description |
|---|---|
| close | Close database connection. |
| execute_query | Execute SQL query and return results with improved error handling. |
| get_connection | Get database connection. |
| validate_credentials | Test if credentials are valid for database connection. |
close
core.database.DatabaseConnection.close()Close database connection.
execute_query
core.database.DatabaseConnection.execute_query(
query,
query_name=None,
params=None,
)Execute SQL query and return results with improved error handling.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| query | Union[str, Path] | SQL query or path to query file | required |
| query_name | Optional[str] | Name of the query, by default None | None |
| params | Optional[dict] | Query parameters, by default None | None |
Returns
| Name | Type | Description |
|---|---|---|
| pd.DataFrame | Query results |
Raises
| Name | Type | Description |
|---|---|---|
| RuntimeError | If query execution fails |
get_connection
core.database.DatabaseConnection.get_connection()Get database connection.
validate_credentials
core.database.DatabaseConnection.validate_credentials(username, password)Test if credentials are valid for database connection.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| username | str | Database username to test | required |
| password | str | Database password to test | required |
Returns
| Name | Type | Description |
|---|---|---|
| bool | True if credentials are valid, False otherwise |