54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
@lru_cache
|
|
def get_engine():
|
|
settings = get_settings()
|
|
return create_engine(
|
|
settings.mysql_dsn,
|
|
pool_pre_ping=True,
|
|
future=True,
|
|
)
|
|
|
|
|
|
SessionLocal = sessionmaker(bind=get_engine(), autoflush=False, autocommit=False, future=True)
|
|
|
|
|
|
def get_db() -> Session:
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def check_database_health() -> dict[str, object]:
|
|
settings = get_settings()
|
|
with get_engine().connect() as conn:
|
|
current_db = conn.execute(text("SELECT DATABASE()")).scalar()
|
|
version = conn.execute(text("SELECT VERSION()")).scalar()
|
|
table_count = conn.execute(
|
|
text(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM information_schema.tables
|
|
WHERE table_schema = :schema_name
|
|
"""
|
|
),
|
|
{"schema_name": settings.mysql_database},
|
|
).scalar()
|
|
return {
|
|
"connected": True,
|
|
"database": current_db,
|
|
"version": version,
|
|
"table_count": int(table_count or 0),
|
|
}
|
|
|