Spaces:
Sleeping
Sleeping
File size: 1,497 Bytes
6c9c901 |
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 |
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from typing import Optional
import logging
from .config import get_settings
logger = logging.getLogger(__name__)
_client: Optional[AsyncIOMotorClient] = None
_db: Optional[AsyncIOMotorDatabase] = None
_connection_healthy = False
async def test_connection() -> bool:
"""Test if the database connection is healthy"""
global _connection_healthy
try:
client = get_client()
# Try to ping the database
await client.admin.command('ping')
_connection_healthy = True
logger.info("Database connection established successfully")
return True
except Exception as e:
_connection_healthy = False
logger.warning(f"Database connection failed: {e}")
return False
def get_client() -> AsyncIOMotorClient:
global _client
if _client is None:
settings = get_settings()
# Use the MongoDB URI exactly as provided, without additional parameters
_client = AsyncIOMotorClient(settings.mongodb_uri)
return _client
def get_database() -> AsyncIOMotorDatabase:
global _db
if _db is None:
client = get_client()
settings = get_settings()
_db = client[settings.database_name]
return _db
def get_collection(name: str):
db = get_database()
return db[name]
def is_database_available() -> bool:
"""Check if database is available for operations"""
return _connection_healthy
|