引言
在现代 Web 开发中，Django 作为 Python 最流行的 Web 框架，以其强大的 ORM 和完善的生态系统著称。而 MongoDB 作为领先的 NoSQL 数据库，以其灵活的文档模型和出色的扩展性受到青睐。将两者结合使用，可以充分发挥 Django 的开发效率和 MongoDB 的数据存储优势。
本文将详细介绍如何在 Django 项目中高效集成 MongoDB，从基础安装配置到高级实战应用，帮助开发者构建高性能的混合数据存储方案。
1. 为什么选择 Django + MongoDB
1.1 优势互补
Django
: 提供完善的 Web 开发框架，包括用户认证、管理后台、表单处理等
MongoDB
: 提供灵活的文档存储，适合处理非结构化数据、大数据量和高并发场景
混合存储
: 可以同时使用关系型数据库（如 PostgreSQL）和 MongoDB，各取所长
1.2 适用场景
需要存储大量非结构化数据（如日志、用户行为数据）
需要快速迭代和灵活的数据模型
需要水平扩展处理高并发读写
需要存储 JSON/XML 等复杂数据结构
2. 环境准备
2.1 安装 MongoDB
2.1.1 Ubuntu/Debian 安装
# 导入 MongoDB 公钥
wget -qO – https://www.mongodb.org/static/pgp/server-7.0.asc | sudo apt-key add –
# 创建列表文件
echo “deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu $(lsb_release -cs)/mongodb-org/7.0 multiverse” | sudo tee /etc/apt/sources.list.d/mongodb-org-7.0.list
# 更新包管理器
sudo apt-get update
# 安装 MongoDB
sudo apt-get install -y mongodb-org
# 启动 MongoDB 服务
sudo systemctl start mongod
sudo systemctl enable mongod
2.1.2 macOS 安装
# 使用 Homebrew 安装
brew tap mongodb/brew
brew install mongodb-community
# 启动服务
brew services start mongodb-community
2.1.3 Docker 安装（推荐）
# 拉取官方镜像
docker pull mongo:7.0
# 启动容器
docker run -d \
–name mongodb \
-p 27017:27017 \
-e MONGO_INITDB_ROOT_USERNAME=admin \
-e MONGO_INITDB_ROOT_PASSWORD=your_password \
-v mongodb_data:/data/db \
mongo:7.0
2.2 创建 Django 项目
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/macOS
# venv\Scripts\activate # Windows
# 安装 Django
pip install django
# 创建项目
django-admin startproject myproject
cd myproject
# 创建应用
python manage.py startapp blog
3. 安装和配置 Django MongoDB 集成库
3.1 选择合适的集成方案
Django 集成 MongoDB 主要有三种方式：
djongo
: 将 MongoDB 伪装成 SQL 数据库，使用 Django ORM
django-mongodb-engine
: 专门为 MongoDB 设计的数据库引擎
直接使用 PyMongo
: 绕过 Django ORM，直接操作 MongoDB
推荐方案
: 对于新项目，推荐使用
djongo
或
直接使用 PyMongo
。本文重点介绍
djongo
方案。
3.2 安装 djongo
pip install djongo
pip install psycopg2 # djongo 依赖
3.3 配置数据库设置
在
settings.py
中配置数据库：
# settings.py
DATABASES = {
‘default’: {
‘ENGINE’: ‘djongo’,
‘NAME’: ‘myproject_db’, # 数据库名称
‘CLIENT’: {
‘host’: ‘localhost’, # MongoDB 主机
‘port’: 27017, # MongoDB 端口
‘username’: ‘admin’, # 用户名（如果设置了认证）
‘password’: ‘your_password’, # 密码
‘authSource’: ‘admin’, # 认证数据库
‘authMechanism’: ‘SCRAM-SHA-1’, # 认证机制
},
‘ENFORCE_SCHEMA’: False, # 是否强制执行模式（推荐 False）
‘LOGGING’: {
‘version’: 1,
‘loggers’: {
‘djongo’: {
‘level’: ‘DEBUG’,
‘propagate’: False,
}
}
}
}
}
3.4 配置数据库别名（可选）
如果需要同时使用多个数据库，可以在
settings.py
中配置：
DATABASES = {
‘default’: {
‘ENGINE’: ‘django.db.backends.postgresql’,
‘NAME’: ‘myproject_db’,
# … PostgreSQL 配置
},
‘mongodb’: {
‘ENGINE’: ‘djongo’,
‘NAME’: ‘myproject_mongo’,
‘CLIENT’: {
‘host’: ‘localhost’,
‘port’: 27017,
}
}
}
4. Django 模型定义
4.1 基础模型定义
使用 djongo 时，模型定义与普通 Django 模型几乎相同：
# blog/models.py
from djongo import models
from django.utils import timezone
class BlogPost(models.Model):
_id = models.ObjectIdField() # MongoDB 的 _id 字段
title = models.CharField(max_length=200)
content = models.TextField()
author = models.CharField(max_length=100)
tags = models.ArrayField(
model_container=models.CharField,
model_form_class=forms.CharField
) # 数组字段
metadata = models.EmbeddedModelField(
model_container=PostMetadata
) # 嵌入式模型
created_at = models.DateTimeField(default=timezone.now)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = ‘blog_posts’ # 自定义集合名称
class PostMetadata(models.Model):
views = models.IntegerField(default=0)
likes = models.IntegerField(default=0)
category = models.CharField(max_length=50)
class Meta:
abstract = True # 抽象模型，不会在数据库中创建集合
4.2 复杂数据结构示例
# blog/models.py
class UserProfile(models.Model):
_id = models.ObjectIdField()
user_id = models.IntegerField(unique=True)
username = models.CharField(max_length=150)
# 嵌入式文档
profile = models.EmbeddedModelField(
model_container=ProfileInfo
)
# 数组嵌入式文档
skills = models.ArrayField(
model_container=Skill
)
# 动态字段（JSONField）
preferences = models.JSONField(default=dict)
class Meta:
db_table = ‘user_profiles’
class ProfileInfo(models.Model):
bio = models.TextField()
avatar = models.URLField()
location = models.CharField(max_length=100)
class Meta:
abstract = True
class Skill(models.Model):
name = models.CharField(max_length=50)
level = models.IntegerField()
class Meta:
abstract = True
4.3 使用抽象基类
# blog/models.py
class BaseModel(models.Model):
_id = models.ObjectIdField()
created_at = models.DateTimeField(default=timezone.now)
updated_at = models.DateTimeField(auto_now=True)
is_active = models.BooleanField(default=True)
class Meta:
abstract = True
class Article(BaseModel):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.CharField(max_length=100)
class Meta:
db_table = ‘articles’
5. 数据库操作实战
5.1 创建数据
# blog/views.py
from django.shortcuts import render
from .models import BlogPost, PostMetadata, UserProfile, Skill
from django.http import JsonResponse
def create_blog_post(request):
# 创建博客文章
post = BlogPost.objects.create(
title=”Django MongoDB 集成指南”,
content=”这是一篇关于 Django 集成 MongoDB 的详细教程…”,
author=”张三”,
tags=[“Django”, “MongoDB”, “Python”],
metadata=PostMetadata(views=100, likes=20, category=”技术教程”)
)
return JsonResponse({
“message”: “创建成功”,
“post_id”: str(post._id)
})
def create_user_profile(request):
# 创建用户资料
profile = UserProfile.objects.create(
user_id=1,
username=”johndoe”,
profile=ProfileInfo(
bio=”Python 开发者”,
avatar=”https://example.com/avatar.jpg”,
location=”北京”
),
skills=[
Skill(name=”Python”, level=90),
Skill(name=”Django”, level=85),
Skill(name=”MongoDB”, level=80)
],
preferences={
“theme”: “dark”,
“language”: “zh-CN”,
“notifications”: True
}
)
return JsonResponse({
“message”: “用户资料创建成功”,
“user_id”: profile.user_id
})
5.2 查询数据
# blog/views.py
def query_posts(request):
# 基础查询
posts = BlogPost.objects.filter(author=”张三”)
# 数组查询（包含特定标签）
posts_with_tag = BlogPost.objects.filter(tags__contains=”Django”)
# 嵌入式文档查询
popular_posts = BlogPost.objects.filter(metadata__views__gt=50)
# 复杂查询（使用 Q 对象）
from django.db.models import Q
complex_posts = BlogPost.objects.filter(
Q(author=”张三”) | Q(metadata__category=”技术教程”)
)
# 聚合查询
from djongo.models import Count
author_stats = BlogPost.objects.values(‘author’).annotate(
post_count=Count(‘_id’)
)
return JsonResponse({
“posts”: list(posts.values()),
“popular_posts”: list(popular_posts.values()),
“author_stats”: list(author_stats)
})
def query_user_profiles(request):
# 查询嵌入式文档字段
beijing_users = UserProfile.objects.filter(profile__location=”北京”)
# 查询数组中的元素
python_developers = UserProfile.objects.filter(skills__name=”Python”)
# 查询 JSON 字段
dark_theme_users = UserProfile.objects.filter(preferences__theme=”dark”)
# 投影查询（只返回特定字段）
users = UserProfile.objects.filter(
username=”johndoe”
).values(‘username’, ‘profile__bio’)
return JsonResponse({
“beijing_users”: list(beijing_users.values()),
“python_developers”: list(python_developers.values()),
“dark_theme_users”: list(dark_theme_users.values()),
“users”: list(users)
})
5.3 更新数据
# blog/views.py
def update_post(request, post_id):
try:
# 单个更新
post = BlogPost.objects.get(_id=post_id)
post.metadata.views += 1
post.save()
# 批量更新
BlogPost.objects.filter(author=”张三”).update(
metadata__views=0
)
# 更新嵌入式文档
BlogPost.objects.filter(_id=post_id).update(
metadata__category=”高级教程”
)
return JsonResponse({“message”: “更新成功”})
except BlogPost.DoesNotExist:
return JsonResponse({“error”: “文章不存在”}, status=404)
def update_user_profile(request, user_id):
# 更新数组字段
profile = UserProfile.objects.get(user_id=user_id)
profile.skills.append(Skill(name=”Redis”, level=75))
profile.save()
# 更新 JSON 字段
UserProfile.objects.filter(user_id=user_id).update(
preferences__language=”en-US”
)
return JsonResponse({“message”: “用户资料更新成功”})
5.4 删除数据
# blog/views.py
def delete_post(request, post_id):
try:
# 单个删除
post = BlogPost.objects.get(_id=post_id)
post.delete()
# 批量删除
BlogPost.objects.filter(author=”张三”).delete()
return JsonResponse({“message”: “删除成功”})
except BlogDoesNotExist:
return JsonResponse({“error”: “文章不存在”}, status=404)
5.5 原始 MongoDB 操作
当 djongo 无法满足复杂需求时，可以直接使用 PyMongo：
# blog/utils.py
from pymongo import MongoClient
from django.conf import settings
def get_mongo_client():
“””获取 MongoDB 客户端”””
client = MongoClient(
host=settings.DATABASES[‘mongodb’][‘CLIENT’][‘host’],
port=settings.DATABASES[‘mongodb’][‘CLIENT’][‘port’]
)
return client
def raw_mongo_operations():
“””原始 MongoDB 操作示例”””
client = get_mongo_client()
db = client.myproject_mongo
collection = db.blog_posts
# 插入文档
result = collection.insert_one({
“title”: “原始 MongoDB 操作”,
“content”: “直接使用 PyMongo”,
“tags”: [“MongoDB”, “PyMongo”],
“metadata”: {
“views”: 50,
“likes”: 10,
“category”: “高级”
}
})
# 复杂聚合查询
pipeline = [
{“$match”: {“author”: “张三”}},
{“$group”: {“_id”: “$metadata.category”, “count”: {“$sum”: 1}}},
{“$sort”: {“count”: -1}}
]
stats = list(collection.aggregate(pipeline))
return {
“inserted_id”: str(result.inserted_id),
“category_stats”: stats
}
6. 高级功能与优化
6.1 索引管理
# blog/models.py
class BlogPost(models.Model):
# … 字段定义
class Meta:
db_table = ‘blog_posts’
indexes = [
models.Index(fields=[‘author’]), # 单字段索引
models.Index(fields=[‘created_at’, ‘-created_at’]), # 降序索引
models.Index(fields=[(‘metadata.views’, 1)]), # 嵌入式字段索引
models.Index(fields=[(‘metadata.category’, 1), (‘created_at’, -1)]), # 复合索引
]
# 手动创建索引（在 Django shell 中）
# python manage.py shell
from blog.models import BlogPost
BlogPost._meta.db_table.create_index([(“metadata.views”, -1)])
6.2 数据库路由与多数据库
# settings.py
DATABASE_ROUTERS = [‘myproject.routers.MongoRouter’]
# myproject/routers.py
class MongoRouter:
def db_for_read(self, model, **hints):
if model._meta.app_label == ‘blog’:
return ‘mongodb’
return None
def db_for_write(self, model, **hints):
if model._meta.app_label == ‘blog’:
return ‘mongodb’
return None
def allow_relation(self, obj1, obj2, **hints):
return True
def allow_migrate(self, db, app_label, model_name=None, **hints):
if app_label == ‘blog’:
return db == ‘mongodb’
return None
6.3 事务处理
# blog/views.py
from djongo import transaction
def atomic_operation(request):
“””原子操作示例”””
try:
with transaction.atomic(using=’mongodb’):
# 在事务中执行多个操作
post = BlogPost.objects.create(
title=”事务测试文章”,
content=”事务内容”,
author=”测试作者”,
metadata=PostMetadata(views=0, likes=0, category=”测试”)
)
# 更新操作
BlogPost.objects.filter(author=”测试作者”).update(
metadata__views=1
)
# 如果这里抛出异常，所有操作都会回滚
# raise Exception(“测试回滚”)
return JsonResponse({“message”: “事务执行成功”})
except Exception as e:
return JsonResponse({“error”: str(e)}, status=500)
6.4 性能优化技巧
6.4.1 批量操作
# 批量插入（比逐个插入快很多）
def bulk_insert_posts(request):
posts = [
BlogPost(
title=f”文章{i}”,
content=f”内容{i}”,
author=”批量作者”,
metadata=PostMetadata(views=i*10, likes=i*2, category=”批量”)
)
for i in100)
]
BlogPost.objects.bulk_create(posts)
return JsonResponse({“message”: “批量插入完成”})
# 批量更新
def bulk_update_posts(request):
BlogPost.objects.filter(author=”批量作者”).update(
metadata__views=1000
)
return JsonResponse({“message”: “批量更新完成”})
6.4.2 只查询需要的字段
# 只查询需要的字段，减少数据传输
def efficient_query(request):
# 不推荐：查询所有字段
# posts = BlogPost.objects.all()
# 推荐：只查询需要的字段
posts = BlogPost.objects.filter(
author=”张三”
).values(‘title’, ‘metadata__views’, ‘created_at’)
return JsonResponse({“posts”: list(posts)})
6.4.3 使用游标处理大量数据
# 处理大量数据时使用游标，避免内存溢出
def process_large_dataset(request):
# 使用 iterator() 避免缓存
for post in BlogPost.objects.filter(
author=”张三”
).iterator(chunk_size=1000):
# 处理每条记录
process_post(post)
return JsonResponse({“message”: “处理完成”})
def process_post(post):
# 业务逻辑处理
pass
7. 实战项目示例
7.1 完整的博客系统
7.1.1 模型定义
# blog/models.py
from djongo import models
from django.utils import timezone
from django import forms
class Comment(models.Model):
user = models.CharField(max_length=100)
content = models.TextField()
created_at = models.DateTimeField(default=timezone.now)
class Meta:
abstract = True
class BlogPost(models.Model):
_id = models.ObjectIdField()
title = models.CharField(max_length=200)
slug = models.SlugField(unique=True)
content = models.TextField()
author = models.CharField(max_length=100)
tags = models.ArrayField(
model_container=models.CharField,
model_form_class=forms.CharField
)
comments = models.ArrayField(
model_container=Comment
)
metadata = models.EmbeddedModelField(
model_container=PostMetadata
)
created_at = models.DateTimeField(default=timezone.now)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = ‘blog_posts’
ordering = [‘-created_at’]
class PostMetadata(models.Model):
views = models.IntegerField(default=0)
likes = models.IntegerField(default=0)
category = models.CharField(max_length=50)
reading_time = models.IntegerField(default=1) # 预计阅读时间（分钟）
class Meta:
abstract = True
7.1.2 视图函数
# blog/views.py
from django.shortcuts import render, get_object_or_404
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from django.views.decorators.csrf import csrf_exempt
import json
from .models import BlogPost, Comment
@require_http_methods([“GET”])
def post_list(request):
“””获取博客文章列表”””
page = int(request.GET.get(‘page’, 1))
page_size = int(request.GET.get(‘page_size’, 10))
category = request.GET.get(‘category’)
# 基础查询
posts_query = BlogPost.objects.all()
# 分类过滤
if category:
posts_query = posts_query.filter(metadata__category=category)
# 分页
offset = (page – 1) * page_size
posts = posts_query[offset:offset + page_size]
# 只返回需要的字段
posts_data = list(posts.values(
‘_id’, ‘title’, ‘slug’, ‘author’, ‘tags’,
‘metadata__views’, ‘metadata__likes’, ‘metadata__category’,
‘created_at’
))
return JsonResponse({
“posts”: posts_data,
“page”: page,
“page_size”: page_size
})
@require_http_methods([“GET”])
def post_detail(request, slug):
“””获取单篇文章详情”””
post = get_object_or_404(BlogPost, slug=slug)
# 增加浏览量（原子操作）
BlogPost.objects.filter(_id=post._id).update(
metadata__views=post.metadata.views + 1
)
return JsonResponse({
“_id”: str(post._id),
“title”: post.title,
“content”: post.content,
“author”: post.author,
“tags”: post.tags,
“comments”: [
{
“user”: c.user,
“content”: c.content,
“created_at”: c.created_at.isoformat()
}
for c in post.comments
],
“metadata”: {
“views”: post.metadata.views + 1,
“likes”: post.metadata.likes,
“category”: post.metadata.category,
“reading_time”: post.metadata.reading_time
},
“created_at”: post.created_at.isoformat(),
“updated_at”: post.updated_at.isoformat()
})
@csrf_exempt
@require_http_methods([“POST”])
def create_post(request):
“””创建新文章”””
data = json.loads(request.body)
post = BlogPost.objects.create(
title=data[‘title’],
slug=data[‘slug’],
content=data[‘content’],
author=data[‘author’],
tags=data.get(‘tags’, []),
metadata=PostMetadata(
views=0,
likes=0,
category=data.get(‘category’, ‘uncategorized’),
reading_time=data.get(‘reading_time’, 1)
)
)
return JsonResponse({
“message”: “创建成功”,
“post_id”: str(post._id),
“slug”: post.slug
}, status=201)
@csrf_exempt
@require_http_methods([“POST”])
def add_comment(request, slug):
“””添加评论”””
post = get_object_or_404(BlogPost, slug=slug)
data = json.loads(request.body)
comment = Comment(
user=data[‘user’],
content=data[‘content’]
)
# 添加到数组
post.comments.append(comment)
post.save()
return JsonResponse({“message”: “评论添加成功”})
@require_http_methods([“GET”])
def get_stats(request):
“””获取统计信息”””
from djongo.models import Count, Sum
# 按分类统计
category_stats = BlogPost.objects.values(
‘metadata__category’
).annotate(
count=Count(‘_id’),
total_views=Sum(‘metadata__views’)
)
# 热门文章（浏览量前10）
popular_posts = BlogPost.objects.order_by(
‘-metadata__views’
)[:10].values(‘title’, ‘metadata__views’, ‘slug’)
return JsonResponse({
“category_stats”: list(category_stats),
“popular_posts”: list(popular_posts)
})
7.1.3 URL 配置
# blog/urls.py
from django.urls import path
from . import views
urlpatterns = [
path(‘posts/’, views.post_list, name=’post_list’),
path(‘posts/
path(‘posts/create/’, views.create_post, name=’create_post’),
path(‘posts/
project/urls.py
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path(‘admin/’, admin.site.urls),
path(‘api/’, include(‘blog.urls’)),
]
7.2 用户行为追踪系统
# analytics/models.py
from djongo import models
from django.utils import timezone
class UserEvent(models.Model):
_id = models.ObjectIdField()
user_id = models.IntegerField()
event_type = models.CharField(max_length=50) # click, view, purchase, etc.
page_url = models.URLField()
timestamp = models.DateTimeField(default=timezone.now)
properties = models.JSONField(default=dict) # 事件属性
class Meta:
db_table = ‘user_events’
indexes = [
models.Index(fields=[‘user_id’]),
models.Index(fields=[‘event_type’]),
models.Index(fields=[‘timestamp’]),
models.Index(fields=[(‘timestamp’, -1), (‘user_id’, 1)]),
]
# analytics/views.py
def track_event(request):
“””追踪用户事件”””
data = json.loads(request.body)
event = UserEvent.objects.create(
user_id=data[‘user_id’],
event_type=data[‘event_type’],
page_url=data[‘page_url’],
properties=data.get(‘properties’, {})
)
return JsonResponse({“status”: “tracked”})
def user_behavior_stats(request, user_id):
“””获取用户行为统计”””
from djongo.models import Count
# 用户事件类型分布
event_stats = UserEvent.objects.filter(
user_id=user_id
).values(‘event_type’).annotate(
count=Count(‘_id’)
)
# 最近活动
recent_events = UserEvent.objects.filter(
user_id=user_id
).order_by(‘-timestamp’)[:20]
return JsonResponse({
“event_stats”: list(event_stats),
“recent_events”: [
{
“type”: e.event_type,
“page”: e.page_url,
“timestamp”: e.timestamp.isoformat(),
“properties”: e.properties
}
for e in recent_events
]
})
8. 测试策略
8.1 单元测试
# blog/tests.py
from django.test import TestCase
from .models import BlogPost, PostMetadata
from djongo import transaction
class BlogPostTest(TestCase):
def setUp(self):
# 测试数据
self.post = BlogPost.objects.create(
title=”测试文章”,
content=”测试内容”,
author=”测试作者”,
tags=[“test”, “django”],
metadata=PostMetadata(views=100, likes=20, category=”测试”)
)
def test_create_post(self):
“””测试创建文章”””
self.assertEqual(self.post.title, “测试文章”)
self.assertEqual(self.post.metadata.views, 100)
def test_query_posts(self):
“””测试查询”””
posts = BlogPost.objects.filter(author=”测试作者”)
self.assertEqual(posts.count(), 1)
# 数组查询
posts_with_tag = BlogPost.objects.filter(tags__contains=”django”)
self.assertEqual(posts_with_tag.count(), 1)
def test_update_post(self):
“””测试更新”””
self.post.metadata.views += 1
self.post.save()
updated = BlogPost.objects.get(_id=self.post._id)
self.assertEqual(updated.metadata.views, 101)
def test_delete_post(self):
“””测试删除”””
post_id = self.post._id
self.post.delete()
with self.assertRaises(BlogPost.DoesNotExist):
BlogPost.objects.get(_id=post_id)
def test_transaction(self):
“””测试事务”””
with transaction.atomic(using=’mongodb’):
post1 = BlogPost.objects.create(
title=”事务文章1″,
content=”内容1″,
author=”事务作者”,
metadata=PostMetadata(views=0, likes=0, category=”事务”)
)
post2 = BlogPost.objects.create(
title=”事务文章2″,
content=”内容2″,
author=”事务作者”,
metadata=PostMetadata(views=0, likes=0, category=”事务”)
)
# 模拟错误
# raise Exception(“测试回滚”)
# 验证两个文章都创建成功
posts = BlogPost.objects.filter(author=”事务作者”)
self.assertEqual(posts.count(), 2)
8.2 集成测试
# blog/tests_integration.py
from django.test import TestCase, Client
from django.urls import reverse
import json
class BlogAPITest(TestCase):
def setUp(self):
self.client = Client()
# 创建测试数据
BlogPost.objects.create(
title=”API测试文章”,
slug=”api-test”,
content=”API测试内容”,
author=”API测试作者”,
tags=[“api”, “test”],
metadata=PostMetadata(views=50, likes=10, category=”API测试”)
)
def test_post_list_api(self):
“””测试文章列表API”””
response = self.client.get(reverse(‘post_list’))
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertIn(‘posts’, data)
self.assertEqual(len(data[‘posts’]), 1)
def test_post_detail_api(self):
“””测试文章详情API”””
response = self.client.get(reverse(‘post_detail’, args=[‘api-test’]))
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertEqual(data[‘title’], “API测试文章”)
def test_create_post_api(self):
“””测试创建文章API”””
data = {
“title”: “新文章”,
“slug”: “new-post”,
“content”: “新内容”,
“author”: “新作者”,
“tags”: [“new”, “api”],
“category”: “API测试”
}
response = self.client.post(
reverse(‘create_post’),
data=json.dumps(data),
content_type=’application/json’
)
self.assertEqual(response.status_code, 201)
# 验证创建成功
post = BlogPost.objects.get(slug=”new-post”)
self.assertEqual(post.title, “新文章”)
9. 部署与生产环境配置
9.1 生产环境数据库配置
# settings.py
import os
# 生产环境配置
DATABASES = {
‘default’: {
‘ENGINE’: ‘djongo’,
‘NAME’: os.getenv(‘MONGO_DB_NAME’, ‘myproject_prod’),
‘CLIENT’: {
‘host’: os.getenv(‘MONGO_HOST’, ‘localhost’),
‘port’: int(os.getenv(‘MONGO_PORT’, 27017)),
‘username’: os.getenv(‘MONGO_USER’, ‘admin’),
‘password’: os.getenv(‘MONGO_PASSWORD’, ”),
‘authSource’: os.getenv(‘MONGO_AUTH_SOURCE’, ‘admin’),
‘authMechanism’: ‘SCRAM-SHA-1’,
‘connectTimeoutMS’: 30000, # 30秒连接超时
‘socketTimeoutMS’: 60000, # 60秒套接字超时
‘maxPoolSize’: 50, # 最大连接池大小
‘minPoolSize’: 5, # 最小连接池大小
},
‘ENFORCE_SCHEMA’: False,
‘LOGGING’: {
‘version’: 1,
‘disable_existing_loggers’: False,
‘formatters’: {
‘verbose’: {
‘format’: ‘{levelname} {asctime} {module} {message}’,
‘style’: ‘{‘,
},
},
‘handlers’: {
‘file’: {
‘level’: ‘DEBUG’,
‘class’: ‘logging.FileHandler’,
‘filename’: ‘mongodb.log’,
‘formatter’: ‘verbose’,
},
},
‘loggers’: {
‘djongo’: {
‘handlers’: [‘file’],
‘level’: ‘DEBUG’,
‘propagate’: False,
},
}
}
}
}
9.2 Docker Compose 部署
# docker-compose.yml
version: ‘3.8’
services:
mongodb:
image: mongo:7.0
container_name: myproject_mongodb
environment:
MONGO_INITDB_ROOT_USERNAME: admin
MONGO_INITDB_ROOT_PASSWORD: ${MONGO_PASSWORD}
MONGO_INITDB_DATABASE: myproject_prod
ports:
– “27017:27017”
volumes:
– mongodb_data:/data/db
– ./init-mongo.js:/docker-entrypoint-initdb.d/init-mongo.js:ro
networks:
– myproject_network
restart: unless-stopped
web:
build: .
container_name: myproject_web
command: gunicorn myproject.wsgi:application –bind 0.0.0.0:8000
environment:
– DJANGO_SETTINGS_MODULE=myproject.settings.production
– MONGO_HOST=mongodb
– MONGO_PORT=27017
– MONGO_USER=admin
– MONGO_PASSWORD=${MONGO_PASSWORD}
– MONGO_DB_NAME=myproject_prod
ports:
– “8000:8000”
depends_on:
– mongodb
networks:
– myproject_network
restart: unless-stopped
nginx:
image: nginx:alpine
container_name: myproject_nginx
ports:
– “80:80”
– “443:443”
volumes:
– ./nginx.conf:/etc/nginx/nginx.conf:ro
– ./ssl:/etc/nginx/ssl:ro
depends_on:
– web
networks:
– myproject_network
restart: unless-stopped
volumes:
mongodb_data:
networks:
myproject_network:
driver: bridge
9.3 MongoDB 初始化脚本
// init-mongo.js
db = db.getSiblingDB(‘myproject_prod’);
// 创建用户
db.createUser({
user: “myproject_user”,
pwd: “your_secure_password”,
roles: [
{ role: “readWrite”, db: “myproject_prod” },
{ role: “dbAdmin”, db: “myproject_prod” }
]
});
// 创建索引
db.blog_posts.createIndex({ “metadata.views”: -1 });
db.blog_posts.createIndex({ “author”: 1 });
db.blog_posts.createIndex({ “created_at”: -1 });
db.blog_posts.createIndex({ “tags”: 1 });
db.blog_posts.createIndex({ “slug”: 1 }, { unique: true });
// 创建用户事件集合索引
db.user_events.createIndex({ “user_id”: 1 });
db.user_events.createIndex({ “event_type”: 1 });
db.user_events.createIndex({ “timestamp”: -1 });
9.4 生产环境监控
# monitoring.py
import logging
from pymongo import MongoClient
from django.conf import settings
logger = logging.getLogger(__name__)
class MongoDBMonitor:
def __init__(self):
self.client = MongoClient(
host=settings.DATABASES[‘default’][‘CLIENT’][‘host’],
port=settings.DATABASES[‘default’][‘CLIENT’][‘port’],
username=settings.DATABASES[‘default’][‘CLIENT’][‘username’],
password=settings.DATABASES[‘default’][‘CLIENT’][‘password’],
serverSelectionTimeoutMS=5000
)
self.db = self.client[settings.DATABASES[‘default’][‘NAME’]]
def check_connection(self):
“””检查数据库连接”””
try:
self.client.admin.command(‘ping’)
return True
except Exception as e:
logger.error(f”MongoDB connection error: {e}”)
return False
def get_collection_stats(self, collection_name):
“””获取集合统计信息”””
stats = self.db.command(‘collstats’, collection_name)
return {
“count”: stats[‘count’],
“size”: stats[‘size’],
“storageSize”: stats[‘storageSize’],
“nindexes”: stats[‘nindexes’],
“totalIndexSize”: stats[‘totalIndexSize’]
}
def get_db_stats(self):
“””获取数据库统计信息”””
stats = self.db.command(‘dbStats’)
return {
“dataSize”: stats[‘dataSize’],
“storageSize”: stats[‘storageSize’],
“objects”: stats[‘objects’],
“collections”: stats[‘collections’]
}
def monitor_slow_queries(self, threshold_ms=100):
“””监控慢查询（需要开启数据库 profiler）”””
slow_queries = self.db.system.profile.find({
“millis”: {“$gte”: threshold_ms}
}).sort(“ts”, -1).limit(10)
return [{
“query”: q.get(‘ns’, ”),
“duration”: q.get(‘millis’, 0),
“timestamp”: q.get(‘ts’)
} for q in slow_queries]
# 使用示例
def health_check(request):
monitor = MongoDBMonitor()
if monitor.check_connection():
stats = monitor.get_db_stats()
return JsonResponse({
“status”: “healthy”,
“database_stats”: stats
})
else:
return JsonResponse({
“status”: “unhealthy”,
“error”: “MongoDB connection failed”
}, status=503)
10. 常见问题与解决方案
10.1 连接问题
问题
: 连接超时或认证失败
解决方案
:
# 增加连接超时时间
‘CLIENT’: {
‘host’: ‘localhost’,
‘connectTimeoutMS’: 30000,
‘socketTimeoutMS’: 60000,
‘serverSelectionTimeoutMS’: 30000,
}
# 检查认证信息
# 确保 authSource 和 authMechanism 配置正确
10.2 性能问题
问题
: 查询速度慢
解决方案
:
# 1. 创建合适的索引
class Meta:
indexes = [
models.Index(fields=[‘author’]),
models.Index(fields=[(‘metadata.views’, -1)]),
]
# 2. 使用投影查询
BlogPost.objects.filter(…).values(‘title’, ‘author’)
# 3. 使用 iterator() 处理大量数据
for post in BlogPost.objects.all().iterator():
process(post)
10.3 数据一致性
问题
: 需要保证数据一致性
解决方案
:
# 使用事务
from djongo import transaction
with transaction.atomic(using=’mongodb’):
# 多个操作
operation1()
operation2()
10.4 模式迁移
问题
: 需要修改模型结构
解决方案
:
# 1. 添加新字段
# 直接在模型中添加，MongoDB 会自动适应
# 2. 重命名字段
# 需要编写数据迁移脚本
def migrate_rename_field():
db = get_mongo_client().myproject_db
db.blog_posts.update_many(
{“old_field”: {“$exists”: True}},
{“$rename”: {“old_field”: “new_field”}}
)
# 3. 删除字段
db.blog_posts.update_many(
{},
{“$unset”: {“deprecated_field”: “”}}
)
11. 最佳实践总结
11.1 设计原则
混合存储
: 核心业务数据使用 PostgreSQL，非结构化数据使用 MongoDB
索引策略
: 为高频查询字段创建索引，避免全表扫描
数据模型
: 合理使用嵌入式文档和引用，避免过度嵌套
查询优化
: 只查询需要的字段，使用投影和分页
11.2 性能优化
批量操作
: 使用
bulk_create()
和
bulk_update()
连接池
: 配置合适的连接池大小
缓存
: 对热点数据使用 Redis 缓存
异步处理
: 使用 Celery 处理耗时操作
11.3 安全考虑
认证授权
: 使用 MongoDB 的角色管理
输入验证
: 验证所有用户输入
SQL 注入
: 虽然 MongoDB 不易受 SQL 注入，但仍需防范注入攻击
备份策略
: 定期备份 MongoDB 数据
11.4 监控与维护
日志记录
: 记录所有数据库操作
性能监控
: 监控查询性能和连接状态
定期清理
: 清理过期数据和日志
版本升级
: 定期更新 MongoDB 和 djongo 版本
12. 总结
Django 集成 MongoDB 是一个强大的组合，能够充分发挥两者的优势。通过本文的详细指导，您应该已经掌握了：
基础配置
: 正确安装和配置 MongoDB 以及 Django 集成
模型设计
: 如何定义复杂的嵌入式和数组字段
数据操作
: CRUD 操作的最佳实践
高级特性
: 索引、事务、多数据库等高级功能
实战应用
: 完整的博客系统和用户行为追踪示例
部署运维
: 生产环境的配置和监控
记住，成功的集成不仅仅是技术实现，更需要考虑数据模型设计、性能优化和运维监控。建议在实际项目中逐步应用这些知识，并根据具体需求进行调整和优化。
如果您在实际使用中遇到任何问题，欢迎参考 MongoDB 官方文档和 djongo 的 GitHub 仓库获取最新信息。