67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
from pymongo import MongoClient, ASCENDING
|
|
from pymongo.errors import ConnectionFailure
|
|
from typing import Optional, List, Dict
|
|
|
|
class MongoDBManager:
|
|
def __init__(self, db_url: str, db_name: str = 'AutoPercenty'):
|
|
"""MongoDB와의 연결을 설정하고 데이터베이스를 설정합니다."""
|
|
try:
|
|
self.client = MongoClient(db_url)
|
|
self.db = self.client[db_name]
|
|
self.collection = None
|
|
print("MongoDB 연결 성공!")
|
|
except ConnectionFailure as e:
|
|
print(f"MongoDB 연결 실패: {e}")
|
|
raise
|
|
|
|
def set_collection(self, collection_name: str):
|
|
"""컬렉션을 설정합니다."""
|
|
self.collection = self.db[collection_name]
|
|
|
|
def insert_one(self, document: Dict) -> bool:
|
|
"""단일 문서를 컬렉션에 추가합니다."""
|
|
try:
|
|
self.collection.insert_one(document)
|
|
return True
|
|
except Exception as e:
|
|
print(f"문서 삽입 실패: {e}")
|
|
return False
|
|
|
|
def find_one(self, query: Dict) -> Optional[Dict]:
|
|
"""단일 문서를 조회합니다."""
|
|
return self.collection.find_one(query)
|
|
|
|
def find(self, query: Dict, projection: Dict = None) -> List[Dict]:
|
|
"""여러 문서를 조회합니다."""
|
|
return list(self.collection.find(query, projection))
|
|
|
|
def update_one(self, query: Dict, update_data: Dict) -> bool:
|
|
"""단일 문서를 업데이트합니다."""
|
|
result = self.collection.update_one(query, {"$set": update_data})
|
|
return result.modified_count > 0
|
|
|
|
def delete_one(self, query: Dict) -> bool:
|
|
"""단일 문서를 삭제합니다."""
|
|
result = self.collection.delete_one(query)
|
|
return result.deleted_count > 0
|
|
|
|
def insert_multiple(self, documents: List[Dict]) -> bool:
|
|
"""여러 문서를 컬렉션에 추가합니다."""
|
|
try:
|
|
if len(documents) > 10:
|
|
documents = documents[:10] # 최대 10개까지만 추가
|
|
self.collection.insert_many(documents)
|
|
return True
|
|
except Exception as e:
|
|
print(f"여러 문서 삽입 실패: {e}")
|
|
return False
|
|
|
|
def find_all(self, query: Dict = {}) -> List[Dict]:
|
|
"""모든 문서를 조회합니다."""
|
|
return list(self.collection.find(query))
|
|
|
|
def close_connection(self):
|
|
"""MongoDB 연결을 닫습니다."""
|
|
self.client.close()
|
|
print("MongoDB 연결이 종료되었습니다.")
|