| """
|
| MongoDB Database Connection Manager
|
| Author: AI Generated
|
| Created: 2025-11-24
|
| Purpose: Handle MongoDB connection and collection access
|
| """
|
|
|
| from pymongo import MongoClient
|
| from pymongo.database import Database
|
| from pymongo.collection import Collection
|
| from config import settings
|
|
|
|
|
| class DatabaseManager:
|
| """Singleton MongoDB connection manager"""
|
|
|
| _instance = None
|
| _client: MongoClient = None
|
| _db: Database = None
|
|
|
| def __new__(cls):
|
| if cls._instance is None:
|
| cls._instance = super().__new__(cls)
|
| return cls._instance
|
|
|
| def __init__(self):
|
| if self._client is None:
|
| self.connect()
|
|
|
| @property
|
| def client(self):
|
| """Get MongoDB client"""
|
| return self._client
|
|
|
| @property
|
| def db_name(self):
|
| """Get database name"""
|
| return settings.DB_NAME
|
|
|
| def connect(self):
|
| """Establish connection to MongoDB"""
|
| self._client = MongoClient(settings.MONGODB_URI)
|
| self._db = self._client[settings.DB_NAME]
|
| print(f"✓ Connected to MongoDB: {settings.DB_NAME}")
|
|
|
| def get_collection(self, collection_name: str) -> Collection:
|
| """Get a MongoDB collection"""
|
| return self._db[collection_name]
|
|
|
| def close(self):
|
| """Close MongoDB connection"""
|
| if self._client:
|
| self._client.close()
|
| print("✓ MongoDB connection closed")
|
|
|
|
|
| @property
|
| def users(self) -> Collection:
|
| """User collection (contains embedded UserFollows array)"""
|
| return self.get_collection(settings.COLLECTION_USERS)
|
|
|
| @property
|
| def payments(self) -> Collection:
|
| """Payment collection"""
|
| return self.get_collection(settings.COLLECTION_PAYMENTS)
|
|
|
| @property
|
| def event_versions(self) -> Collection:
|
| """EventVersion collection"""
|
| return self.get_collection(settings.COLLECTION_EVENT_VERSIONS)
|
|
|
| @property
|
| def post_social_media(self) -> Collection:
|
| """PostSocialMedia collection (contains nested Images.UserCommentPosts)"""
|
| return self.get_collection(settings.COLLECTION_POST_SOCIAL_MEDIA)
|
|
|
|
|
| @property
|
| def audience_segments(self) -> Collection:
|
| """AudienceSegment collection (DEPRECATED - use event_audience_segments)"""
|
| return self.get_collection(settings.COLLECTION_AUDIENCE_SEGMENTS)
|
|
|
| @property
|
| def user_segment_assignments(self) -> Collection:
|
| """UserSegmentAssignment collection"""
|
| return self.get_collection(settings.COLLECTION_USER_SEGMENT_ASSIGNMENTS)
|
|
|
| @property
|
| def sentiment_results(self) -> Collection:
|
| """SentimentAnalysisResult collection"""
|
| return self.get_collection(settings.COLLECTION_SENTIMENT_RESULTS)
|
|
|
| @property
|
| def event_insights(self) -> Collection:
|
| """EventInsightReport collection"""
|
| return self.get_collection(settings.COLLECTION_EVENT_INSIGHTS)
|
|
|
|
|
| @property
|
| def event_audience_segments(self) -> Collection:
|
| """EventAudienceSegment collection"""
|
| return self.get_collection("EventAudienceSegment")
|
|
|
| @property
|
| def event_sentiment_summary(self) -> Collection:
|
| """EventSentimentSummary collection"""
|
| return self.get_collection("EventSentimentSummary")
|
|
|
|
|
|
|
| db = DatabaseManager()
|
|
|
