copywriting_production/src/tools/backup_mongo.py

import asyncio
from datetime import datetime
from typing import Optional, List
from beanie import Document, Indexed, init_beanie
from beanie.odm.operators.update.general import Set
from pydantic import BaseModel
from motor.motor_asyncio import AsyncIOMotorClient
from config.settings import MONGO_URL, MONGO_DB_NAME
from src.models.product_model import Product
from src.manager.core.db_mongo import BaseMongoManager
from utils.logu import get_logger
from upath import UPath
logger = get_logger('tools')

async def test_product_mongo():
    db_mongo = BaseMongoManager()
    await db_mongo.initialize()
    product = await Product.find_one(Product.basic_info.name == "电线保护套")
async def main():
   await test_product_mongo()

if __name__ == "__main__":
    asyncio.run(main())