将 MongoEngine 与 Flask 集成非常简单。核心思想是利用 MongoEngine 自带的 connect() 函数，在 Flask 应用初始化（通常是使用应用工厂模式）时建立与 MongoDB 的连接，之后便可以直接在蓝图或视图函数中定义模型（Document）并进行 CRUD 操作。
# api.py
from mongoengine import connect, NotUniqueError, ValidationError
from dbconfig import Config
from pymongo import MongoClient
def create_app(config_class=Config):
app = Flask(__name__)
CORS(app, resources=r'/*')
app.config.from_object(config_class)
# --- 核心：原生初始化 MongoEngine 连接 ---
# 从配置中解包字典并连接数据库
# 1. 初始化 MongoEngine（供 models.py 使用）
connect(**app.config['MONGODB_SETTINGS'])
# (选配)2. 初始化原生 PyMongo 客户端，并挂载到 app 上
app.mongo_client = MongoClient(app.config['MONGO_URI'])
# 获取具体的数据库对象（从 URI 中自动解析库名，或写死 db = app.mongo_client['flask_demo_db']）
app.db = app.mongo_client.get_default_database()
# 注册蓝图
app.register_blueprint(main_bp, url_prefix='/main')
app.register_blueprint(agent_bp, url_prefix='/agent')
return app
# dbconfig
class Config:
# MongoEngine 的连接配置字典
MONGODB_SETTINGS = {
'db': 'graph',
'host': '127.0.0.1',
'port': 27017,
'username': 'jacky',
'password': password,
'authentication_source': 'admin'
}
# (选配) 原生 PyMongo 连接 URI
MONGO_URI = f'mongodb://jacky:{password}@127.0.0.1:27017/graph?authSource=admin'
# modelRoute.py
from mongoengine import NotUniqueError, ValidationError
from models import Hiuser
# 创建一个名为 'main' 的蓝图
main_bp = Blueprint('main', __name__)
@main_bp.route('/getusers', methods=['GET'])
def get_users():
"""获取所有用户"""
users = Hiuser.objects()
return jsonify([user.to_dict() for user in users]), 200
# models.py
from datetime import datetime, timedelta
# from mongoengine import Document, StringField, EmailField, DateTimeField
from mongoengine import (
Document, StringField, IntField, EmailField, DateTimeField,
ListField, EmbeddedDocument, EmbeddedDocumentField, NotUniqueError
)
class Hiuser(Document):
name = StringField(required=True, max_length=50)
email = EmailField(required=True, unique=True)
# created_at = DateTimeField(default=datetime.utcnow)
created_at = DateTimeField(default=lambda: datetime.now(timezone.utc), required=True)
# 映射到 MongoDB 中的集合名称（可选，默认使用类名的小写复数形式）
meta = {'collection': 'hiusers'}
def to_dict(self):
"""辅助方法：将 Document 转换为字典，方便 JSON 序列化"""
return {
"id": str(self.id),
"name": self.name,
"email": self.email,
"created_at": self.created_at.isoformat()
}
# agentRoute.py
# 这里将 MongoDBSaver 作为 checkpointer
from flask import Blueprint, jsonify, current_app
from langchain_core.messages import message_to_dict
from deepagents import create_deep_agent
from langgraph.checkpoint.mongodb import MongoDBSaver
# 创建一个名为 'agent' 的蓝图
agent_bp = Blueprint('agent', __name__)
@agent_bp.route('/gethistory', methods=['GET'])
def gethistory():
# 通过 current_app 获取挂载的数据库对象
db = current_app.db
collection = db["chatdbs"]
checkpointer = MongoDBSaver(collection)
agent = create_deep_agent(
model=llm,
tools=[internet_search],
system_prompt=research_instructions,
checkpointer=checkpointer
)
config = {"configurable": {"thread_id": "conversation-001"}}
# 遍历该线程的所有历史状态
snapshots = agent.get_state_history(config)
try:
temp = next(snapshots)
chats = temp.values['messages']
# 将每条消息转换为字典
serialized_messages = [message_to_dict(chat) for chat in chats]
return jsonify(serialized_messages), 200
except Exception as e:
print(44, "error", e)
return "error", 500
共享连接池：虽然 MongoEngine 和 PyMongo 分别初始化，但它们底层都由 PyMongo 的 MongoClient 驱动，并且共享底层的 Socket 连接池，不会造成资源浪费。
- 需要严格校验、字段类型约束、面向对象操作的业务（如用户系统、权限管理），继续使用 models.py 里的 MongoEngine。
- 需要极高性能、灵活多变、无需预定义 Schema 的业务（如埋点日志、动态属性表、聚合管道聚合查询），直接在其他路由里用 PyMongo (current_app.db.collection)。