49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from app.db.engine import engine
|
|
from app.db.session import SessionLocal
|
|
from app.db.models import *
|
|
from app.core.config import settings
|
|
|
|
def init_db():
|
|
try:
|
|
if settings.DEBUG:
|
|
print("Creating database tables...")
|
|
Base.metadata.create_all(bind=engine)
|
|
if settings.DEBUG:
|
|
print("Database tables created.")
|
|
except Exception as e:
|
|
print(f"Error creating database tables: {e}")
|
|
raise
|
|
finally:
|
|
if settings.DEBUG:
|
|
print("Tables :", Base.metadata.tables.keys())
|
|
|
|
|
|
def clear_db():
|
|
try:
|
|
if settings.DEBUG:
|
|
print("Deleting database tables...")
|
|
Base.metadata.drop_all(bind=engine)
|
|
if settings.DEBUG:
|
|
print("Database tables deleted.")
|
|
except Exception as e:
|
|
print(f"Error deleting database tables: {e}")
|
|
raise
|
|
|
|
def reset_db():
|
|
session = SessionLocal()
|
|
try:
|
|
if settings.DEBUG:
|
|
print("Deleting database rows...")
|
|
|
|
for table in reversed(Base.metadata.sorted_tables):
|
|
session.execute(table.delete())
|
|
session.commit()
|
|
|
|
if settings.DEBUG:
|
|
print("Database rows deleted.")
|
|
except Exception as e:
|
|
print(f"Error deleting database rows: {e}")
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|