Python + MongoDB 实战：快速搭建数据管理API服务

🚀 项目概述

今天我们来分享一个使用Python FastAPI框架搭配MongoDB数据库的实战项目。这个项目能够快速搭建一个完整的数据管理API服务，非常适合学习后端开发或者需要快速原型验证的场景。

📦 技术栈版本

  • Python: 12版本
  • MongoDB: 8.2.3
  • PyMongo: ≥4.15.5
  • FastAPI: 现代高性能Web框架
  • Uvicorn: ASGI服务器

🏗️ 核心架构

🛠️ 完整项目代码

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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
from datetime import datetime
from typing import Optional
import uvicorn

app = FastAPI(title="MongoDB测试", version="1.0.0")

# MongoDB连接（直接写URL，方便测试）
MONGO_URL = "mongodb://admin:secret@172.19.199.196:27017"
DB_NAME = "test_db"

# 初始化MongoDB连接
client = MongoClient(MONGO_URL)
db = client[DB_NAME]
collection = db["items"]

# 测试页面
@app.get("/")
async def home():
    return {
        "message": "MongoDB测试服务",
        "endpoints": {
            "创建数据": "POST /create?name=xxx&description=yyy",
            "查看所有": "GET /all",
            "查看一个": "GET /item/{id}",
            "搜索": "GET /search?name=xxx",
            "删除": "DELETE /delete/{id}"
        }
    }

# 创建数据（GET方式，方便浏览器测试）
@app.get("/create")
async def create_item(name: str, description: Optional[str] = None):
    """创建一条测试数据"""
    item = {
        "name": name,
        "description": description,
        "created_at": datetime.utcnow()
    }
    
    result = collection.insert_one(item)
    item["_id"] = str(result.inserted_id)
    return {"message": "创建成功", "data": item}

# 获取所有数据
@app.get("/all")
async def get_all():
    """获取所有数据"""
    items = []
    for item in collection.find().limit(20):
        item["_id"] = str(item["_id"])  # 转换ObjectId为字符串
        items.append(item)
    
    return {
        "count": len(items),
        "data": items
    }

# 根据ID获取数据
@app.get("/item/{item_id}")
async def get_item(item_id: str):
    """根据ID获取数据"""
    from bson import ObjectId
    
    try:
        item = collection.find_one({"_id": ObjectId(item_id)})
    except:
        return {"error": "ID格式不正确"}
    
    if not item:
        return {"error": "未找到数据"}
    
    item["_id"] = str(item["_id"])
    return item

# 搜索数据
@app.get("/search")
async def search_items(name: str):
    """根据名称搜索数据"""
    items = []
    for item in collection.find({"name": {"$regex": name, "$options": "i"}}):
        item["_id"] = str(item["_id"])
        items.append(item)
    
    return {
        "count": len(items),
        "data": items
    }

# 删除数据
@app.delete("/delete/{item_id}")
async def delete_item(item_id: str):
    """删除数据"""
    from bson import ObjectId
    
    try:
        result = collection.delete_one({"_id": ObjectId(item_id)})
    except:
        return {"error": "ID格式不正确"}
    
    if result.deleted_count == 0:
        return {"error": "未找到数据"}
    
    return {"message": "删除成功", "deleted_count": result.deleted_count}

# 测试数据库连接
@app.get("/test-connection")
async def test_connection():
    """测试MongoDB连接"""
    try:
        # 发送一个ping命令测试连接
        client.admin.command('ping')
        return {"status": "连接成功", "database": DB_NAME}
    except Exception as e:
        return {"status": "连接失败", "error": str(e)}

# 清空测试数据（小心使用）
@app.get("/clear")
async def clear_all():
    """清空所有测试数据"""
    result = collection.delete_many({})
    return {"message": f"已删除 {result.deleted_count} 条数据"}

if __name__ == "__main__":
    print("🚀 启动MongoDB测试服务...")
    print("📊 数据库:", DB_NAME)
    print("🔗 MongoDB URL:", MONGO_URL)
    print("🌐 访问地址: http://localhost:8000")
    print("📚 API文档: http://localhost:8000/docs")
    
    uvicorn.run(app, host="0.0.0.0", port=8000)

1. 数据库连接配置

1
2
3
4
5
# 简洁明了的MongoDB连接配置
MONGO_URL = "mongodb://admin:secret@172.19.199.196:27017"
DB_NAME = "test_db"
client = MongoClient(MONGO_URL)
db = client[DB_NAME]

记忆点: MongoDB连接字符串格式为 mongodb://用户名:密码@主机:端口

2. 数据模型设计

项目使用灵活的文档模型，每条数据包含：

  • name: 名称（必填）
  • description: 描述（可选）
  • created_at: 创建时间（自动生成）
  • _id: 唯一标识（MongoDB自动生成）

🔧 API接口详解

📍 首页导航 (GET /)

提供所有可用接口的说明，方便开发者快速了解API功能。

➕ 创建数据 (GET /create)

特色: 使用GET方式创建数据，方便浏览器直接测试

1
2
3
4
5
6
# 自动添加创建时间戳
item = {
    "name": name,
    "description": description,
    "created_at": datetime.utcnow()  # 使用UTC时间
}

📋 查询数据

  • GET /all: 获取所有数据（限制20条，防止数据过大）
  • GET /item/{id}: 根据ID精确查询
  • GET /search?name=xxx: 根据名称模糊搜索（不区分大小写）

🗑️ 删除数据 (DELETE /delete/{id})

提供安全的数据删除功能，返回明确的删除结果。

🧪 实用工具接口

  • GET /test-connection: 测试数据库连接状态
  • GET /clear: 清空测试数据（慎用！）

💡 关键技术点

1. ObjectId处理

1
2
3
# ObjectId需要特殊处理才能序列化为JSON
from bson import ObjectId
item["_id"] = str(item["_id"])  # 转换为字符串

2. 模糊搜索实现

1
2
# 使用正则表达式实现不区分大小写的搜索
{"name": {"$regex": name, "$options": "i"}}

3. 错误处理

1
2
3
4
try:
    item = collection.find_one({"_id": ObjectId(item_id)})
except:
    return {"error": "ID格式不正确"}

🚦 快速启动

运行项目只需要一行命令：

1
python app.py

启动后可以访问：

🎯 项目特色

  1. 即开即用: 无需复杂配置，修改MongoDB连接信息即可使用
  2. 便于测试: 使用GET方法进行创建操作，浏览器直接访问
  3. 完整功能: CRUD操作齐全，满足基本数据管理需求
  4. 良好实践: 包含错误处理、连接测试等生产级特性

📚 学习价值

通过这个项目，你可以学习到：

  • FastAPI的基本使用
  • MongoDB与Python的集成
  • RESTful API设计
  • 数据库连接管理
  • 错误处理和输入验证

🔍 扩展建议

如需在生产环境使用，建议：

  1. 将GET创建改为POST，符合REST规范
  2. 添加输入验证（使用Pydantic）
  3. 增加分页功能
  4. 添加用户认证
  5. 完善日志记录

总结

这个项目展示了如何用最少的代码构建一个功能完整的API服务。MongoDB的灵活文档模型加上FastAPI的高效特性，让后端开发变得简单快捷。无论是学习还是快速验证想法，这都是一个很好的起点。

技术要点回顾：

  • MongoDB 8.2.3 + PyMongo 4.15.5
  • FastAPI自动生成API文档
  • 灵活的文档数据库操作
  • 完整的CRUD实现

赶紧动手试试吧，在自己的电脑上跑起来感受一下！ 🎉