MongoDB数据库操作封装

.gitignore

@@ -0,0 +1,4 @@
+.idea/
+.vscode/
+__pycache__/
+*.pyc

config/__init__.py

@@ -0,0 +1,4 @@
+from config.config import load_config
+__all__ = [
+    "load_config"
+]

config/config.py

@@ -0,0 +1,6 @@
+import yaml
+
+def load_config():
+    with open('./config/config.yaml', 'r') as file:
+        config = yaml.safe_load(file)
+    return config

config/config.yaml

@@ -0,0 +1,6 @@
+mongodb:
+  host: "qa-general.andunip.cn"
+  port: 2817
+  username: "AndunAI"  # 如果需要身份验证
+  password: "Andun@AI$2025*"  # 如果需要身份验证
+  database: "ai-monitor"

db/__init__.py

@@ -0,0 +1,4 @@
+from db.db_client import MongoClientHelper
+__all__ = [
+    "MongoClientHelper"
+]

db/dao/dao.py

@@ -0,0 +1,48 @@
+from db import MongoClientHelper
+
+class MongoDao:
+    _instance = None
+    
+    def __new__(cls, collection_name):
+        if not cls._instance:
+            cls._instance = super(MongoDao, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+    
+    def __init__(self, collection_name):
+        if not self._initialized:
+            self.db_client = MongoClientHelper()
+            self._initialized = True
+        self.collection_name = collection_name
+            
+    def get_one_record_by_query(self, query):
+        res = self.db_client.find_one(self.collection_name, query)
+        return res
+    
+    def get_records_by_query(self, query):
+        collections = self.db_client.find_many(self.collection_name, query)
+        records = [collection for collection in collections]
+        return records
+    
+    def get_one_field_data(self, field):
+        fields = [field]
+        """获取指定key的所有数据，返回列表"""
+        field_records = self.db_client.find_fields(self.collection_name, fields)
+        return [record[field] for record in field_records]
+    
+    def get_fields_data(self, fields):
+        records = self.db_client.find_fields(self.collection_name, fields)
+        return [record for record in records]
+    
+    def get_all_records(self):
+        records = self.db_client.find_all(self.collection_name)
+        return [record for record in records]
+
+        
+
+if __name__ == '__main__':
+    collection_name = "obrand-ec"
+    dao = MongoDao(collection_name)
+    field = "nick"
+    res = dao.get_all_records()
+    print(len(res))

db/db_client.py

@@ -0,0 +1,65 @@
+from config import load_config
+from pymongo import MongoClient
+from pymongo.errors import ConnectionFailure
+from urllib.parse import quote_plus
+import pandas as pd
+
+class MongoClientHelper:
+    def __init__(self):
+        self.cfgs = load_config()['mongodb']
+        self.client = None
+        self.db = None
+        
+        if self.client is None:
+            self.connect()
+        
+    def connect(self):
+        """连接MongoDB"""
+        try:
+            host = self.cfgs['host']
+            port = self.cfgs['port']
+            username = quote_plus(self.cfgs['username'])
+            password = quote_plus(self.cfgs['password'])
+            database = self.cfgs['database']
+            
+            uri = f"mongodb://{username}:{password}@{host}:{port}/{database}?authSource=admin"
+            self.client = MongoClient(uri)
+            self.db = self.client[database]
+            
+        except ConnectionFailure as e:
+            print(f"MongoDB数据库连接失败: {e}")
+            
+    def find_one(self, collection_name, query):
+        """根据条件查询一条结果"""
+        collection = self.db[collection_name]
+        return collection.find_one(query)
+    
+    def find_many(self, collection_name, query):
+        """根据条件查询多条结果"""
+        collection = self.db[collection_name]
+        return collection.find(query)
+    
+    def find_all(self, collection_name):
+        """返回mongdb中指定collection中的所有数据"""
+        collection = self.db[collection_name]
+        return collection.find()
+    
+    def find_fields(self, collection_name, fields):
+        """获取指定字段的所有数据"""
+        collection = self.db[collection_name]
+        projection = {field: 1 for field in fields}
+        projection["_id"] = 0
+        return collection.find({}, projection)
+            
+if __name__ == "__main__":
+    db_client = MongoClientHelper()
+    collection_name = 'obrand-ec'
+    query = {"nick": "李宁篮球旗舰店"}
+    fields = ["nick"]
+    res = db_client.find_fields(collection_name, fields)
+    l = []
+    for r in res:
+        l.append(r)
+    data = pd.DataFrame(l)
+    print(data)
+