This repository was archived by the owner on Jul 19, 2025. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathmanage_database.py
More file actions
executable file
Β·259 lines (215 loc) Β· 8.6 KB
/
manage_database.py
File metadata and controls
executable file
Β·259 lines (215 loc) Β· 8.6 KB
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
#!/usr/bin/env python3
"""
Database Management Script
This script provides utilities for managing the Telugu corpus database:
- Check database initialization status
- Force database reinitialization (if needed)
- Show database statistics
- Clean up database volumes
"""
import os
import sys
import argparse
import logging
import subprocess
from pathlib import Path
# Add the app directory to the Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from app.core.config import settings
from setup_postgresql import (
test_postgres_connection,
test_database_connection,
seed_initial_data
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def check_database_status():
"""Check the current status of the database."""
print("π Checking Database Status")
print("=" * 50)
# Check PostgreSQL connection
if test_postgres_connection():
print("β
PostgreSQL server is accessible")
else:
print("β PostgreSQL server is not accessible")
return False
# Check database connection
if test_database_connection():
print("β
Target database is accessible")
else:
print("β Target database is not accessible")
return False
# Check if database has data
try:
from app.models import Role, User, Category, Record
from app.db.session import engine
from sqlmodel import Session, select, text
with Session(engine) as session:
# Check roles
roles_count = len(session.exec(select(Role)).all())
print(f"π Roles: {roles_count}")
# Check users
users_count = len(session.exec(select(User)).all())
print(f"π₯ Users: {users_count}")
# Check categories
categories_count = len(session.exec(select(Category)).all())
print(f"π Categories: {categories_count}")
# Check records
records_count = len(session.exec(select(Record)).all())
print(f"π Records: {records_count}")
# Check PostGIS status
try:
result = session.exec(text("SELECT PostGIS_Version()")).first()
if result:
print(f"πΊοΈ PostGIS Version: {result[0]}")
else:
print("β PostGIS not available")
except Exception:
print("β PostGIS not enabled")
# Determine initialization status
if roles_count >= 3:
print("\nβ
Database appears to be fully initialized")
return True
else:
print("\nβ οΈ Database appears to be partially initialized or empty")
return False
except Exception as e:
print(f"β Error checking database contents: {e}")
return False
def show_database_info():
"""Show detailed database information."""
print("π Database Information")
print("=" * 50)
print(f"π§ Configuration:")
print(f" Host: {settings.DB_HOST}")
print(f" Port: {settings.DB_PORT}")
print(f" Database: {settings.DB_NAME}")
print(f" User: {settings.DB_USER}")
print(f" URL: {settings.DATABASE_URL}")
return check_database_status()
def force_reinitialize():
"""Force database reinitialization (WARNING: This will destroy existing data)."""
print("β οΈ FORCE REINITIALIZATION")
print("=" * 50)
print("WARNING: This will destroy all existing data in the database!")
confirm = input("Are you sure you want to continue? Type 'YES' to confirm: ")
if confirm != 'YES':
print("β Operation cancelled")
return False
try:
# Run the full initialization process
from init_db import initialize_database
print("π Starting forced reinitialization...")
# Drop all tables first
print("ποΈ Dropping all tables...")
try:
result = subprocess.run(["alembic", "downgrade", "base"],
capture_output=True, text=True, cwd=".")
if result.returncode == 0:
print("β
Tables dropped successfully")
else:
print(f"β οΈ Warning during table drop: {result.stderr}")
except Exception as e:
print(f"β οΈ Could not drop tables: {e}")
# Run full initialization
if initialize_database():
print("β
Database reinitialization completed successfully!")
return True
else:
print("β Database reinitialization failed!")
return False
except Exception as e:
print(f"β Error during reinitialization: {e}")
return False
def clean_docker_volumes():
"""Clean up Docker volumes (WARNING: This will destroy all data)."""
print("π§Ή CLEAN DOCKER VOLUMES")
print("=" * 50)
print("WARNING: This will destroy all Docker volumes and data!")
confirm = input("Are you sure you want to continue? Type 'YES' to confirm: ")
if confirm != 'YES':
print("β Operation cancelled")
return False
try:
print("π Stopping Docker containers...")
subprocess.run(["docker-compose", "down"], cwd=".")
print("ποΈ Removing Docker volumes...")
subprocess.run(["docker", "volume", "rm", "corpus-te_postgres_data",
"corpus-te_redis_data", "corpus-te_celery_beat_data"],
capture_output=True)
print("β
Docker volumes cleaned. You can now restart with fresh data.")
print(" Run: docker-compose up -d")
return True
except Exception as e:
print(f"β Error cleaning volumes: {e}")
return False
def seed_sample_data():
"""Seed sample data for testing."""
print("π± Seeding Sample Data")
print("=" * 50)
try:
# Check if database is initialized
if not check_database_status():
print("β Database is not properly initialized. Please run database initialization first.")
return False
# Run the test data creation script
script_path = Path(__file__).parent / "create_test_users.py"
if script_path.exists():
print("π₯ Creating test users...")
result = subprocess.run([sys.executable, str(script_path)],
capture_output=True, text=True, cwd=".")
if result.returncode == 0:
print("β
Test users created successfully")
print(result.stdout)
else:
print(f"β Failed to create test users: {result.stderr}")
return False
else:
print("β οΈ Test user creation script not found")
# Run the test data creation script if available
test_data_script = Path(__file__).parent / "tests" / "create_test_data.py"
if test_data_script.exists():
print("π Creating test data...")
result = subprocess.run([sys.executable, str(test_data_script), "create"],
capture_output=True, text=True, cwd=".")
if result.returncode == 0:
print("β
Test data created successfully")
print(result.stdout)
else:
print(f"β οΈ Test data creation had issues: {result.stderr}")
return True
except Exception as e:
print(f"β Error seeding sample data: {e}")
return False
def main():
"""Main function."""
parser = argparse.ArgumentParser(description="Database Management Utilities")
parser.add_argument("action", choices=[
"status", "info", "reinit", "clean", "seed"
], help="Action to perform")
args = parser.parse_args()
print("ποΈ Telugu Corpus Database Management")
print("=" * 60)
if args.action == "status":
success = check_database_status()
elif args.action == "info":
success = show_database_info()
elif args.action == "reinit":
success = force_reinitialize()
elif args.action == "clean":
success = clean_docker_volumes()
elif args.action == "seed":
success = seed_sample_data()
else:
print("β Unknown action")
success = False
if success:
sys.exit(0)
else:
sys.exit(1)
if __name__ == "__main__":
main()