| | """
|
| | 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()
|
| |
|
