MongoDB 與 Async Python：現代應用開發的強大組合

from beanie import Document, Indexed
from pydantic import BaseModel
from typing import List

# 定義嵌入式檔案
class Category(BaseModel):
    name: str
    description: str

# 定義主檔案模型
class Product(Document):
    name: str
    description: str
    price: Indexed(float)  # 建立索引以加速查詢
    categories: List[Category]  # 嵌入式檔案列表
    
    class Settings:
        name = "products"  # 集合名稱

# 查詢範例
async def find_cheap_chocolates():
    chocolates = await Product.find(
        Product.categories.name == "chocolate",
        Product.price < 5.0
    ).to_list()
    return chocolates

from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

class User(BaseModel):
    id: str
    username: str
    email: str
    full_name: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.now)
    tags: List[str] = []

# 使用模型
user_data = {
    "id": "user123",
    "username": "pythonista",
    "email": "user@example.com",
    "tags": ["python", "mongodb"]
}
user = User(**user_data)
print(user.model_dump_json(indent=2))

import asyncio

async def fetch_data(delay, name):
    print(f"開始取得 {name} 資料...")
    await asyncio.sleep(delay)  # 模擬I/O操作
    print(f"{name} 資料取得完成")
    return f"{name} 結果"

async def main():
    # 平行執行多個協程
    results = await asyncio.gather(
        fetch_data(2, "使用者"),
        fetch_data(1, "產品"),
        fetch_data(3, "訂單")
    )
    print(f"所有結果: {results}")

# 執行非同步程式
asyncio.run(main())

開始取得 使用者 資料...
開始取得 產品 資料...
開始取得 訂單 資料...
產品 資料取得完成
使用者 資料取得完成
訂單 資料取得完成
所有結果: ['使用者 結果', '產品 結果', '訂單 結果']

pip install beanie motor pydantic

import asyncio
from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient

async def init_db():
    # 建立Motor客戶端
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    
    # 初始化Beanie
    await init_beanie(
        database=client.my_database,
        document_models=[
            # 這裡將放置我們的檔案模型類別
        ]
    )

# 在應用啟動時執行
asyncio.run(init_db())

from beanie import Document, Indexed
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

# 嵌入式檔案模型
class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

# 主檔案模型
class User(Document):
    name: str
    email: Indexed(str, unique=True)  # 建立唯一索引
    age: Optional[int] = None
    addresses: List[Address] = []
    created_at: datetime = datetime.now()
    
    class Settings:
        name = "users"  # 集合名稱

async def demo_crud():
    # 建立使用者
    user = User(
        name="玄貓",
        email="blackcat@example.com",
        age=30,
        addresses=[
            Address(
                street="科技路 123 號",
                city="台北",
                country="台灣",
                postal_code="10001"
            )
        ]
    )
    
    # 儲存到資料函式庫
    await user.insert()
    print(f"已建立使用者: {user.id}")
    
    # 查詢使用者
    found_user = await User.find_one(User.email == "blackcat@example.com")
    print(f"找到使用者: {found_user.name}")
    
    # 更新使用者
    found_user.age = 31
    await found_user.save()
    print("使用者已更新")
    
    # 刪除使用者
    await found_user.delete()
    print("使用者已刪除")

# 執行範例
asyncio.run(demo_crud())

async def advanced_queries():
    # 插入一些測試資料
    await User(name="Teresa", email="teresa@example.com", age=25).insert()
    await User(name="Rita", email="rita@example.com", age=30).insert()
    await User(name="BlackCat", email="blackcat@example.com", age=35).insert()
    
    # 基本查詢
    users = await User.find(User.age >= 30).to_list()
    print(f"30歲以上的使用者: {[user.name for user in users]}")
    
    # 排序
    users = await User.find().sort(User.age).to_list()
    print(f"按年齡排序的使用者: {[user.name for user in users]}")
    
    # 限制結果數量
    users = await User.find().limit(2).to_list()
    print(f"限制結果的使用者: {[user.name for user in users]}")
    
    # 複合條件
    users = await User.find(
        User.age > 20,
        User.name.match("^[AB]")  # 名字以A或B開頭
    ).to_list()
    print(f"複合條件的使用者: {[user.name for user in users]}")
    
    # 聚合查詢
    avg_age = await User.find().aggregate([
        {"$group": {"_id": None, "avg_age": {"$avg": "$age"}}}
    ]).to_list()
    print(f"平均年齡: {avg_age[0]['avg_age']}")

# 執行範例
asyncio.run(advanced_queries())

users:
- id
- name
- email

posts:
- id
- title
- content
- user_id (外部索引鍵)

comments:
- id
- content
- post_id (外部索引鍵)
- user_id (外部索引鍵)

class Comment(BaseModel):
    content: str
    author_name: str
    created_at: datetime = datetime.now()

class Post(Document):
    title: str
    content: str
    author: User  # 嵌入完整使用者檔案
    comments: List[Comment] = []  # 嵌入評論
    created_at: datetime = datetime.now()

class User(Document):
    name: str
    email: str

class Comment(Document):
    content: str
    post_id: PydanticObjectId  # 參照帖子ID
    user_id: PydanticObjectId  # 參照使用者ID
    created_at: datetime = datetime.now()

class Post(Document):
    title: str
    content: str
    user_id: PydanticObjectId  # 參照使用者ID
    created_at: datetime = datetime.now()

class CommentSummary(BaseModel):
    id: PydanticObjectId
    content_preview: str  # 評論內容預覽
    author_name: str

class Post(Document):
    title: str
    content: str
    user_id: PydanticObjectId  # 參照使用者ID
    author_name: str  # 冗餘儲存作者名稱，避免查詢
    comment_count: int = 0  # 評論計數
    recent_comments: List[CommentSummary] = []  # 最近評論的摘要
    created_at: datetime = datetime.now()

from beanie import Document, Indexed, Link
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

# 嵌入式檔案
class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str
    is_default: bool = False

class ProductVariant(BaseModel):
    name: str
    sku: str
    price: float
    stock: int

# 主檔案
class User(Document):
    name: str
    email: Indexed(str, unique=True)
    addresses: List[Address] = []
    created_at: datetime = datetime.now()
    
    class Settings:
        name = "users"

class Product(Document):
    name: str
    description: str
    base_price: float
    category: str
    variants: List[ProductVariant] = []
    created_at: datetime = datetime.now()
    
    class Settings:
        name = "products"

class OrderItem(BaseModel):
    product_id: PydanticObjectId
    product_name: str  # 冗餘儲存
    variant_name: str
    price: float  # 購買時的價格
    quantity: int

class Order(Document):
    user_id: PydanticObjectId
    user_email: str  # 冗餘儲存
    items: List[OrderItem]
    total_amount: float
    shipping_address: Address
    status: str = "pending"
    created_at: datetime = datetime.now()
    
    class Settings:
        name = "orders"

from beanie import Document, Indexed, Link
from pydantic import BaseModel, HttpUrl
from typing import List, Optional, Dict
from datetime import datetime

# 嵌入式檔案
class ReleaseFile(BaseModel):
    filename: str
    url: HttpUrl
    size: int
    md5_digest: str
    sha256_digest: str
    upload_time: datetime
    python_version: str
    requires_python: Optional[str] = None
    packagetype: str  # 'sdist', 'bdist_wheel' 等

class Release(BaseModel):
    version: str
    released_at: datetime
    summary: Optional[str] = None
    description: Optional[str] = None
    requires_dist: List[str] = []
    files: List[ReleaseFile] = []
    yanked: bool = False
    yanked_reason: Optional[str] = None

class ProjectUrl(BaseModel):
    name: str  # 例如 'Homepage', 'Documentation', 'Source'
    url: HttpUrl

# 主檔案
class User(Document):
    username: Indexed(str, unique=True)
    email: Optional[str] = None
    name: Optional[str] = None
    joined_at: datetime = datetime.now()
    
    class Settings:
        name = "users"

class Package(Document):
    name: Indexed(str, unique=True)
    normalized_name: Indexed(str, unique=True)  # 用於不區分大小寫的查詢
    summary: Optional[str] = None
    description: Optional[str] = None
    author: Optional[str] = None
    author_email: Optional[str] = None
    maintainers: List[str] = []  # 維護者使用者名列表
    project_urls: List[ProjectUrl] = []
    license: Optional[str] = None
    keywords: List[str] = []
    classifiers: List[str] = []
    releases: Dict[str, Release] = {}  # 版本號 -> 發布訊息
    latest_version: Optional[str] = None
    created_at: datetime = datetime.now()
    last_updated: datetime = datetime.now()
    download_count: int = 0
    
    class Settings:
        name = "packages"
        indexes = [
            "keywords",
            "classifiers",
            "maintainers",
            [("normalized_name", "text"), ("summary", "text"), ("description", "text")]  # 全文索引
        ]

class Download(Document):
    package_name: str
    version: str
    file_name: str
    timestamp: datetime = datetime.now()
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    country_code: Optional[str] = None
    
    class Settings:
        name = "downloads"
        indexes = [
            "package_name",
            "version",
            "timestamp"
        ]

async def pypi_queries():
    # 查詢特定套件
    requests = await Package.find_one(Package.normalized_name == "requests")
    
    # 取得最新版本訊息
    latest_version = requests.latest_version
    latest_release = requests.releases.get(latest_version)
    
    # 查詢特定維護者的所有套件
    kennethreitz_packages = await Package.find(
        Package.maintainers.contains("kennethreitz")
    ).to_list()
    
    # 全文搜尋
    http_packages = await Package.find(
        {"$text": {"$search": "http client"}}
    ).sort([("$text", {"$meta": "textScore"})]).to_list()
    
    # 查詢最受歡迎的套件（按下載量）
    popular_packages = await Package.find().sort(-Package.download_count).limit(10).to_list()
    
    # 查詢最近更新的套件
    recent_packages = await Package.find().sort(-Package.last_updated).limit(10).to_list()
    
    # 查詢特定分類別的套件
    web_frameworks = await Package.find(
        Package.classifiers.contains("Framework :: Django")
    ).to_list()

async def add_new_release(package_name: str, version: str, release_data: dict):
    package = await Package.find_one(Package.normalized_name == package_name.lower())
    
    if not package:
        # 建立新套件
        package = Package(
            name=package_name,
            normalized_name=package_name.lower(),
            # 其他欄位...
        )
    
    # 增加新版本
    release = Release(
        version=version,
        released_at=datetime.now(),
        # 從release_data填充其他欄位...
    )
    
    package.releases[version] = release
    
    # 更新最新版本（如果適用）
    if not package.latest_version or version > package.latest_version:
        package.latest_version = version
    
    package.last_updated = datetime.now()
    
    await package.save()

async def record_download(package_name: str, version: str, file_name: str, request_info: dict):
    # 記錄下載
    download = Download(
        package_name=package_name,
        version=version,
        file_name=file_name,
        ip_address=request_info.get("ip"),
        user_agent=request_info.get("user_agent"),
        country_code=request_info.get("country")
    )
    await download.insert()
    
    # 更新套件下載計數（可以使用原子更新）
    await Package.find_one(
        Package.normalized_name == package_name.lower()
    ).update({"$inc": {"download_count": 1}})