File size: 3,600 Bytes
d97be90 ea06065 d97be90 ea06065 d97be90 ea06065 d97be90 ea06065 d97be90 ea06065 d97be90 ea06065 d97be90 |
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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
"""
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")
# ACTUAL Collections (matching models.txt)
@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)
# AI Result Collections
@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)
# NEW: Event-centric collections
@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")
# Global database instance
db = DatabaseManager()
|