-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathverify_db.py
37 lines (29 loc) · 1.29 KB
/
verify_db.py
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
import os
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from config import DevelopmentConfig, ProductionConfig
def verify_db():
if os.environ.get('APP_MODE') == 'production':
config = ProductionConfig()
else:
config = DevelopmentConfig()
engine = create_engine(config.SQLALCHEMY_DATABASE_URI)
Session = sessionmaker(bind=engine)
session = Session()
try:
crime_data = session.execute(text("SELECT COUNT(*) FROM crime_data")).fetchone()
print(f"crime_data table has {crime_data[0]} records.")
unfounded_data = session.execute(text("SELECT COUNT(*) FROM unfounded_data")).fetchone()
print(f"unfounded_data table has {unfounded_data[0]} records.")
meta_data = session.execute(text("SELECT COUNT(*) FROM meta_data")).fetchone()
print(f"meta_data table has {meta_data[0]} records.")
meta = session.execute(text("SELECT * FROM meta_data")).fetchone()
print(f"meta_data: {meta}")
raw_data = session.execute(text("SELECT COUNT(*) FROM raw_data")).fetchone()
print(f"raw_data table has {raw_data[0]} records.")
except Exception as e:
print(f"Error verifying database: {e}")
finally:
session.close()
if __name__ == "__main__":
verify_db()