File size: 3,131 Bytes
9d92c17 48ae4e0 9d92c17 48ae4e0 9d92c17 073826a 48ae4e0 9d92c17 073826a 9d92c17 073826a 9d92c17 073826a 9d92c17 073826a 9d92c17 073826a |
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 |
"""Database package exports.
This package exposes both the new SQLAlchemy-based ``DatabaseManager`` and the
legacy SQLite-backed ``Database`` class that the existing application modules
still import via ``from database import Database``. During the transition phase
we dynamically load the legacy implementation from the root ``database.py``
module (renamed here as ``legacy_database`` when importing) and fall back to the
new manager if that module is unavailable.
"""
from importlib import util as _importlib_util
from pathlib import Path as _Path
from typing import Optional as _Optional, Any as _Any
from .db_manager import DatabaseManager
def _load_legacy_module():
"""Load the legacy root-level ``database.py`` module if it exists.
This is used to support older entry points like ``get_database`` and the
``Database`` class that live in the legacy file.
"""
legacy_path = _Path(__file__).resolve().parent.parent / "database.py"
if not legacy_path.exists():
return None
spec = _importlib_util.spec_from_file_location("legacy_database", legacy_path)
if spec is None or spec.loader is None:
return None
module = _importlib_util.module_from_spec(spec)
try:
spec.loader.exec_module(module) # type: ignore[union-attr]
except Exception:
# If loading the legacy module fails we silently fall back to DatabaseManager
return None
return module
def _load_legacy_database_class() -> _Optional[type]:
"""Load the legacy ``Database`` class from ``database.py`` if available."""
module = _load_legacy_module()
if module is None:
return None
return getattr(module, "Database", None)
def _load_legacy_get_database() -> _Optional[callable]:
"""Load the legacy ``get_database`` function from ``database.py`` if available."""
module = _load_legacy_module()
if module is None:
return None
return getattr(module, "get_database", None)
_LegacyDatabase = _load_legacy_database_class()
_LegacyGetDatabase = _load_legacy_get_database()
_db_manager_instance: _Optional[DatabaseManager] = None
if _LegacyDatabase is not None:
Database = _LegacyDatabase
else:
Database = DatabaseManager
def get_database(*args: _Any, **kwargs: _Any) -> _Any:
"""Return a database instance compatible with legacy callers.
The resolution order is:
1. If the legacy ``database.py`` file exists and exposes ``get_database``,
use that function (this returns the legacy singleton used by the
Gradio crypto dashboard and other older modules).
2. Otherwise, return a singleton instance of ``DatabaseManager`` from the
new SQLAlchemy-backed implementation.
"""
if _LegacyGetDatabase is not None:
return _LegacyGetDatabase(*args, **kwargs)
global _db_manager_instance
if _db_manager_instance is None:
_db_manager_instance = DatabaseManager()
# Ensure tables are created for the monitoring schema
_db_manager_instance.init_database()
return _db_manager_instance
__all__ = ["DatabaseManager", "Database", "get_database"]
|