Spaces:
Sleeping
Sleeping
| import sqlite3 | |
| import os | |
| DB_DIR = "db" | |
| def fetch_db_rows_as_dicts(db_path, table_name): | |
| conn = None | |
| try: | |
| # Connect to the SQLite database | |
| dbPath = os.path.abspath(os.path.join(os.getcwd(), DB_DIR,db_path)) | |
| conn = sqlite3.connect(dbPath) | |
| 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] | |
| # Execute a query to fetch all rows from the table | |
| cursor.execute(f"SELECT * FROM {table_name};") | |
| rows = cursor.fetchall() | |
| assert len(rows) > 1 | |
| return column_names, rows[1:] | |
| except sqlite3.Error as e: | |
| #print(f"SQLite error: {e}") | |
| pass | |
| finally: | |
| # Close the connection | |
| if conn: | |
| conn.close() | |
| # Example usage: | |
| #dbPath = os.path.abspath(os.path.join(os.getcwd(), DB_DIR,'topologies.sqlite')) | |
| #fetch_db_rows_as_dicts(dbPath, 'topologies') | |