这里是级联删除的通用Python实现。我们假设“外键”是父对象的ObjectId。
enter方法在进入上下文时被调用,并自动调用discover_collections方法来查找具有ObjectId引用的集合。 exit方法在退出上下文时被调用,并关闭MongoDB客户端连接。
使用该类作为上下文管理器可确保正确管理MongoDB连接,并在适当的时间调用discover_collections方法。
请记得将<mongodb_connection_string>替换为您实际的MongoDB连接字符串,将“your_database_name”替换为您的数据库名称。
在上下文中调用该类:
with CascadeDelete('<mongodb_connection_string>', 'your_database_name') as cascade_delete:
cascade_delete.delete(ObjectId('parent1'))
CascadeDelete类的实现:
from pymongo import MongoClient
from bson.objectid import ObjectId
class CascadeDelete:
""" Usage:
with CascadeDelete('<mongodb_connection_string>', 'your_database_name') as cascade_delete:
cascade_delete.delete(ObjectId('parent1'))
"""
def __init__(self, connection_string, database_name):
self.client = MongoClient(connection_string)
self.db = self.client[database_name]
self.collections = []
def __enter__(self):
self.discover_collections()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.client.close()
def discover_collections(self):
for collection_name in self.db.list_collection_names():
collection = self.db[collection_name]
if self._has_objectid_references(collection):
self.collections.append(collection)
def delete(self, document_id):
deleted_documents = set()
self._delete_documents(document_id, deleted_documents)
for collection in self.collections:
collection.delete_many({'_id': {'$in': list(deleted_documents)}})
def _delete_documents(self, document_id, deleted_documents):
deleted_documents.add(document_id)
for collection in self.collections:
document = collection.find_one({'_id': document_id})
if document:
for key, value in document.items():
if isinstance(value, ObjectId) and key != '_id':
self._delete_documents(value, deleted_documents)
@staticmethod
def _has_objectid_references(collection):
sample_document = collection.find_one()
if not sample_document:
return False
for value in sample_document.values():
if isinstance(value, ObjectId):
return True
return False
我希望这能帮助到某人...
此致敬礼