Spaces:
Running
Running
File size: 2,856 Bytes
73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 091f243 6224e10 73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 73c06df 6224e10 1a64680 6224e10 73c06df 6224e10 73c06df 6224e10 091f243 6224e10 | 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 | import os
import logging
from urllib.parse import parse_qs, urlparse
from dotenv import load_dotenv
from pymongo import MongoClient
from datetime import datetime
from pydantic import BaseModel
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
load_dotenv()
mongodb_uris = os.getenv('MONGODB_URIS')
# ---------- Pydantic Response Model ----------
class MongoPingResponse(BaseModel):
service_name: str
success: bool
error: str | None = None
time: str
class MongoPingAllResponse(BaseModel):
mongodb_services: list[MongoPingResponse]
# ---------- Helpers ----------
def extract_app_name_from_uri(uri: str) -> str:
"""Extracts the appName from MongoDB URI if present."""
try:
parsed_uri = urlparse(uri)
query_params = parse_qs(parsed_uri.query)
app_name = query_params.get('appName', ['MongoDB Ping Service'])[0]
return app_name
except Exception as e:
logger.debug(f"Failed to extract appName from URI: {e}")
return "MongoDB Ping Service" # Fallback if parsing fails
# ---------- MongoDB Ping ----------
def ping_mongodb() -> list[MongoPingResponse]:
"""Ping MongoDB and return structured responses."""
responses: list[MongoPingResponse] = []
if not mongodb_uris:
logger.warning("No MongoDB URIs found in environment variables")
return responses
for uri in mongodb_uris.split(','):
uri = uri.strip()
if not uri:
continue
service_name = extract_app_name_from_uri(uri)
now = datetime.utcnow().isoformat()
try:
client = MongoClient(uri, serverSelectionTimeoutMS=3000) # 3s timeout
client.admin.command('ping')
logger.info(f"{service_name}: Successfully pinged MongoDB at {uri[:50]}...")
responses.append(
MongoPingResponse(
service_name=service_name,
success=True,
error=None,
time=now
)
)
client.close()
except Exception as e:
logger.error(f"{service_name}: Error pinging MongoDB at {uri[:50]}... - {str(e)}")
responses.append(
MongoPingResponse(
service_name=service_name,
success=False,
error=str(e),
time=now
)
)
return responses
def ping_mongodb_projects():
logger.info("Starting MongoDB ping checks...")
results = ping_mongodb()
logger.info("Completed MongoDB ping checks.")
final_response = MongoPingAllResponse(mongodb_services=results)
return final_response
|