如何使用FastAPI预填充MongoDB数据库？

对于MongoDB这类非关系型数据库，不需要依赖类似Alembic的迁移工具，以下是几种实用的预填充数据方案：

1. 利用FastAPI启动事件自动填充 #

在应用启动时检查目标集合是否为空，若为空则插入预设数据，适合开发环境或需要确保每次启动都有基础数据的场景。

示例代码（基于Motor异步客户端）：

from fastapi import FastAPI
from pydantic import BaseModel
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()

# 定义Pydantic模型
class Item(BaseModel):
    name: str
    description: str | None = None

# MongoDB连接配置
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client["your_db_name"]
items_collection = db["items"]

# 预设初始数据
DEFAULT_ITEMS = [
    {"name": "Sample Item 1", "description": "First default item"},
    {"name": "Sample Item 2", "description": "Second default item"}
]

# 启动时初始化数据
@app.on_event("startup")
async def init_database():
    # 检查集合是否已有数据
    document_count = await items_collection.count_documents({})
    if document_count == 0:
        await items_collection.insert_many(DEFAULT_ITEMS)

# 测试接口
@app.get("/items")
async def get_items():
    items = await items_collection.find().to_list(100)
    return items

2. 使用MongoDB官方导入工具mongoimport #

将预设数据保存为JSON/CSV文件，通过mongoimport命令在应用启动前批量导入，适合生产环境（避免代码硬编码数据）。

示例命令：

# 导入JSON数组格式的数据文件
mongoimport --db your_db_name --collection items --file default_items.json --jsonArray

可以把该命令加入应用启动脚本（如Dockerfile的CMD、shell启动脚本），确保应用启动前完成数据导入。

3. 自定义CLI工具手动触发填充 #

用Typer（FastAPI内置的CLI框架）或Click编写命令行工具，手动执行数据填充，适合需要按需更新初始数据的场景。

示例代码：

import typer
import asyncio
from pydantic import BaseModel
from motor.motor_asyncio import AsyncIOMotorClient

app = typer.Typer()

class Item(BaseModel):
    name: str
    description: str | None = None

DEFAULT_ITEMS = [
    {"name": "Sample Item 1", "description": "First default item"},
    {"name": "Sample Item 2", "description": "Second default item"}
]

@app.command()
def seed_database():
    """填充MongoDB初始数据"""
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    db = client["your_db_name"]
    items_collection = db["items"]

    async def insert_data():
        count = await items_collection.count_documents({})
        if count == 0:
            await items_collection.insert_many(DEFAULT_ITEMS)
            typer.echo("初始数据填充完成")
        else:
            typer.echo("数据库已有数据，跳过填充")

    asyncio.run(insert_data())

if __name__ == "__main__":
    app()

运行命令执行填充：

python seed_script.py seed-database

内容的提问来源于stack exchange，提问作者AfonsoSalgadoSousa