Spaces:
Running
Running
File size: 2,998 Bytes
7011b92 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
import os
import pandas as pd
from sqlalchemy import create_engine, text
# --- CONFIGURATION ---
DB_NAME = "mshauri_fedha.db"
DB_CONNECTION = f"sqlite:///{DB_NAME}"
def list_all_tables(engine):
print(f"\n --- DATABASE SUMMARY: {DB_NAME} ---")
try:
with engine.connect() as conn:
# Query the master table for all table names
query = text("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;")
tables = conn.execute(query).fetchall()
if not tables:
print(" Database is empty.")
return []
table_list = [t[0] for t in tables]
print(f"{'ID':<4} | {'Rows':<8} | {'Table Name'}")
print("-" * 60)
for i, t_name in enumerate(table_list):
# Count rows for verification
try:
count = conn.execute(text(f'SELECT COUNT(*) FROM "{t_name}"')).scalar()
print(f"{i:<4} | {count:<8} | {t_name}")
except:
print(f"{i:<4} | {'ERROR':<8} | {t_name}")
return table_list
except Exception as e:
print(f" Connection failed: {e}")
return []
def inspect_table(engine, table_name):
print(f"\n🔎 Inspecting Table: '{table_name}'")
try:
# Read schema/columns
query = f'SELECT * FROM "{table_name}" LIMIT 5'
df = pd.read_sql(query, engine)
if df.empty:
print(" Table is empty.")
else:
print(f"Columns: {list(df.columns)}")
print("\n--- First 5 Rows ---")
# to_string() makes it readable in terminal without truncation
print(df.to_string(index=False))
print("-" * 50)
except Exception as e:
print(f" Could not read table: {e}")
def main():
if not os.path.exists(DB_NAME):
print(f" Error: Database file '{DB_NAME}' not found in current directory.")
print(f"Current Directory: {os.getcwd()}")
return
engine = create_engine(DB_CONNECTION)
tables = list_all_tables(engine)
if not tables: return
while True:
try:
user_input = input("\nEnter Table ID (or Name) to inspect, or 'q' to quit: ").strip()
if user_input.lower() == 'q': break
target_table = None
# Handle numeric ID input
if user_input.isdigit():
idx = int(user_input)
if 0 <= idx < len(tables):
target_table = tables[idx]
# Handle name input
elif user_input in tables:
target_table = user_input
if target_table:
inspect_table(engine, target_table)
else:
print(" Invalid selection.")
except KeyboardInterrupt:
break
if __name__ == "__main__":
main() |