当前位置：首页 > news >正文

FastAPI删除mongodb重复数据（数据清洗）

news 2025/8/14 7:18:08

在 FastAPI 中删除 MongoDB 重复数据，你需要结合使用 MongoDB 查询和 FastAPI 的路由功能。以下是一个通用的例子，演示如何删除特定字段上的重复数据：

1. 定义数据模型:

from pydantic import BaseModel, Field
from bson import ObjectId
from typing import Optionalclass PyObjectId(ObjectId):@classmethoddef __get_validators__(cls):yield cls.validate@classmethoddef validate(cls, v):if not ObjectId.is_valid(v):raise ValueError("Invalid objectid")return ObjectId(v)@classmethoddef __modify_schema__(cls, field_schema):field_schema.update(type="string")class ItemBase(BaseModel):field_to_check: str # 需要检查重复的字段class Item(ItemBase):id: Optional[PyObjectId] = Field(alias="_id")class Config:arbitrary_types_allowed = Truejson_encoders = {ObjectId: str}

2. 创建 MongoDB 连接:

from motor.motor_asyncio import AsyncIOMotorClientMONGO_DETAILS = "mongodb://localhost:27017" # 替换为你的 MongoDB 连接字符串
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client["your_database_name"] # 替换为你的数据库名称
collection = database.get_collection("your_collection_name") # 替换为你的集合名称

3. 实现删除逻辑:

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()@app.delete("/items/duplicates/", response_model=list[Item])
async def delete_duplicate_items(field_name: str = "field_to_check"):"""删除指定字段上的重复数据。Args:field_name (str, optional): 需要检查重复的字段名. Defaults to "field_to_check".Returns:list[Item]: 返回删除的重复文档列表."""# 使用聚合管道查找并删除重复项pipeline = [{"$match": {"version": 1}},  # 只处理 version 为 1 的文档{"$group": {"_id": {"{}".format(field_name): "$"+field_name}, "count": {"$sum": 1}, "dups": {"$push": "$_id"}}},{"$match": {"count": {"$gt": 1}}},{"$unwind": "$dups"},{"$skip": 1}, {"$project": {"_id": "$dups"}}]duplicate_ids = [doc["_id"] async for doc in collection.aggregate(pipeline)]if duplicate_ids:deleted_items = []for item_id in duplicate_ids:result = await collection.find_one_and_delete({"_id": item_id})if result:deleted_items.append(Item(**result))return deleted_itemsraise HTTPException(status_code=404, detail="没有找到重复数据")

4. 运行 FastAPI 应用:

uvicorn main:app --reload

解释:

数据模型: 使用 Pydantic 定义数据模型，确保数据一致性.
MongoDB 连接: 使用 motor 库异步连接到 MongoDB 数据库.
聚合管道: 使用 MongoDB 的聚合管道查找重复数据：
- $group: 按指定字段分组，计算每个分组中文档数量.
- $match: 筛选数量大于 1 的分组，即存在重复数据的组.
- $unwind: 将 dups 数组展开为多行.
- $skip: 跳过每组的第一个文档，因为我们只删除重复的.
- $project: 只保留 _id 字段.
删除数据: 使用 find_one_and_delete 方法删除找到的重复文档.
错误处理: 如果没有找到重复数据，抛出 404 错误.