copywriting_production/tests/mytest/t_mongo.py

from datetime import datetime
from pydantic import BaseModel
from typing import List, Dict, Optional,TypedDict
from pymongo import MongoClient

uri = "mongodb://sv-v2:27017/"
client = MongoClient(uri)
database = client.get_database("amazone")
def create_collection():
    # https://www.mongodb.com/zh-cn/docs/languages/python/pymongo-driver/current/databases-collections/
    create_collection = database.create_collection("test")
    print(create_collection)
    for i in range(10):
        create_collection.insert_one({"name": f"test{i}"})

def show_collection():
    collection = database.get_collection("test")
    for i in collection.find():
        print(i)
        print(type(i))

class Variant(TypedDict):
    name: str
    price: float
    description: str
    category: str

class Product(BaseModel):
    name: Optional[str] = None
    price: Optional[float]
    # description: str
    description: Optional[str] = None
    image: Optional[str] = None
    brand: Optional[str] = None
    category: Optional[str] = None
    variant: Optional[List[Variant]] = None


def test_get_product():
    product = Product(
        name="数据线",
        price=100,
        description="产品描述", 
        image='123123',
        variant=[
           Variant(
                name="数据线",
                price=100,
                description="产品描述",
                category="电子产品"
            ), 
        ]

    )
    return product
def insert_object():
    product = test_get_product()
    collection = database.get_collection("test")
    collection.insert_one(product)

def find_one():
    collection = database.get_collection("test")
    results = collection.find_one({ "name" : "数据线" })
    print(results)
    print(results["name"])
    results = collection.find_one({ "price" : 100 })
    print(results)

def update_one():
    collection = database.get_collection("test")
    results = collection.find_one({ "name" : "数据线" })
    results_model = Product(**results)
    results_model.image = "123"
    # results = collection.update_one({ "name" : "数据线" }, { "$set": { "image": "123" } })
    collection.update_one(upsert=True,filter={ "name" : "数据线" }, update={ "$set": { "image": "123" } })
    print(results)

class Version(TypedDict):
    desc: str
    create_time: datetime
    version: str


def backup_document():
    back_collection = database['backup']
    product = test_get_product()
    version = Version(
       desc="完成 MongoDB 数据库操作的测试和备份",
       create_time=datetime.now(), 
    )
    product['version'] = version
    back_collection.insert_one(product)
def main():
    # create_collection()
    # show_collection()
    # insert_object()
    # find_one()
    update_one()
    # backup_document()

if __name__ == "__main__":
    main()