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

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

1. from pydantic import BaseModel, Field
2. from bson import ObjectId
3. from typing import Optional
5. class PyObjectId(ObjectId):
6.     @classmethod
7.     def __get_validators__(cls):
8.         yield cls.validate
10.     @classmethod
11.     def validate(cls, v):
12.         if not ObjectId.is_valid(v):
13.             raise ValueError("Invalid objectid")
14.         return ObjectId(v)
16.     @classmethod
17.     def __modify_schema__(cls, field_schema):
18.         field_schema.update(type="string")
20. class ItemBase(BaseModel):
21.     field_to_check: str # 需要检查重复的字段
23. class Item(ItemBase):
24.     id: Optional[PyObjectId] = Field(alias="_id")
26.     class Config:
27.         arbitrary_types_allowed = True
28.         json_encoders = {ObjectId: str}

复制代码

2. 创建 MongoDB 连接:

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

复制代码

3. 实现删除逻辑:

1. from fastapi import FastAPI, HTTPException
3. app = FastAPI()
5. @app.delete("/items/duplicates/", response_model=list[Item])
6. async def delete_duplicate_items(field_name: str = "field_to_check"):
7.     """
8.     删除指定字段上的重复数据。
10.     Args:
11.         field_name (str, optional): 需要检查重复的字段名. Defaults to "field_to_check".
13.     Returns:
14.         list[Item]: 返回删除的重复文档列表.
15.     """
16.     # 使用聚合管道查找并删除重复项
17.     pipeline = [
18.         {"$match": {"version": 1}},  # 只处理 version 为 1 的文档
19.         {"$group": {"_id": {"{}".format(field_name): "$"+field_name}, "count": {"$sum": 1}, "dups": {"$push": "$_id"}}},
20.         {"$match": {"count": {"$gt": 1}}},
21.         {"$unwind": "$dups"},
22.         {"$skip": 1},
23.         {"$project": {"_id": "$dups"}}
24.     ]
26.     duplicate_ids = [doc["_id"] async for doc in collection.aggregate(pipeline)]
27.     if duplicate_ids:
28.         deleted_items = []
29.         for item_id in duplicate_ids:
30.             result = await collection.find_one_and_delete({"_id": item_id})
31.             if result:
32.                 deleted_items.append(Item(**result))
33.         return deleted_items
35.     raise HTTPException(status_code=404, detail="没有找到重复数据")

复制代码

4. 运行 FastAPI 应用:

1. uvicorn main:app --reload

复制代码

解释:

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

注意:

• 将代码中的占位符替换为你自己的数据库和聚集名称.
  
• 可以根据需要修改聚合管道，以适应差别的重复数据查找需求.
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。