Spaces:
Sleeping
Sleeping
| import sqlite3 | |
| def fetch_db_rows_as_dicts(db_path, table_name): | |
| try: | |
| # Connect to the SQLite database | |
| conn = sqlite3.connect(db_path) | |
| conn.row_factory = sqlite3.Row # This allows us to access columns by name | |
| cursor = conn.cursor() | |
| # Get the column names | |
| cursor.execute(f"PRAGMA table_info({table_name});") | |
| columns_info = cursor.fetchall() | |
| column_names = [col[1] for col in columns_info] | |
| # Print the column names | |
| print("Column names:", column_names) | |
| # Execute a query to fetch all rows from the table | |
| cursor.execute(f"SELECT * FROM {table_name};") | |
| rows = cursor.fetchall() | |
| #for row in rows: | |
| # row_dict = dict(zip(column_names, row)) | |
| # print(row_dict) | |
| assert len(rows) > 1 | |
| return column_names, rows[1:] | |
| except sqlite3.Error as e: | |
| print(f"SQLite error: {e}") | |
| finally: | |
| # Close the connection | |
| if conn: | |
| conn.close() | |
| # Example usage: | |
| fetch_db_rows_as_dicts('topologies.sqlite', 'topologies') | |