我是如何建构系统的？——从 “边界思维” 到 “系统性构建”：编程学习的终极方法论（万字版）

# 测试FastAPI的性能
from fastapi import FastAPI

app = FastAPI()

@app.post("/order/create")
def create_order(data: dict):
    # 模拟订单创建逻辑
    return {"code": 201, "message": "订单创建成功"}

# 运行服务：uvicorn test:app --workers 4
# 用wrk测试：wrk -t4 -c100 -d30s http://localhost:8000/order/create -s post.lua
# 预期结果：QPS≥5000

┌─────────────────────────────────────────────────────────┐
│ 表现层（Presentation Layer）                            │
│ - FastAPI接口                                      │
│ - 统一请求/响应格式                                    │
│ - 权限认证、限流、日志                                  │
└─────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────┐
│ 业务层（Business Layer）                                │
│ - 订单服务：创建、支付、查询                          │
│ - 支付服务：调用微信/支付宝接口                          │
│ - 通知服务：发送短信/邮件通知                          │
│ - 业务规则：订单状态流转、库存扣减                          │
└─────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────┐
│ 数据层（Data Layer）                                     │
│ - PostgreSQL：订单、用户、商品数据                          │
│ - Redis：订单缓存、库存缓存                              │
│ - 数据访问对象（DAO）：封装数据库操作                          │
└─────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────┐
│ 基础层（Infrastructure Layer）                         │
│ - 消息队列：异步处理订单通知、统计                          │
│ - 工具函数：时间处理、加密、验证                          │
│ - 配置中心：统一管理系统配置                          │
└─────────────────────────────────────────────────────────┘

# 订单模块核心接口（边界定义）
class OrderService:
    # 创建订单：参数必须包含用户ID、商品ID、数量、金额；返回订单ID
    def create_order(self, user_id: str, product_id: str, quantity: int, amount: float) -> dict:
        pass
    
    # 支付订单：参数必须包含订单ID、支付渠道；返回支付结果
    def pay_order(self, order_id: str, channel: str) -> dict:
    
    # 查询订单：参数必须包含订单ID；返回订单详情
    def get_order(self, order_id: str) -> dict:

# 支付模块核心接口（边界定义）
class PaymentService:
    # 调用支付接口：参数必须包含订单ID、金额、渠道；返回支付结果
    def call_payment_api(self, order_id: str, amount: float, channel: str) -> dict:
    
    # 处理支付回调：参数必须包含支付结果；返回处理结果
    def handle_payment_callback(self, data: dict) -> dict:

# 成功返回
{
    "code": 200,
    "message": "成功",
    "data": {"order_id": "20240525_ORDER_001", "status": "pending"}
}

# 失败返回
{
    "code": 400,
    "message": "订单金额必须≥0",
    "data": {}
}

# 订单服务核心代码
from app.repositories.order_repo import OrderRepo
from app.repositories.product_repo import ProductRepo
from app.services.payment_service import PaymentService

class OrderService:
    def __init__(self):
        self.order_repo = OrderRepo()
        self.product_repo = ProductRepo()
        self.payment_service = PaymentService()
    
    def create_order(self, user_id, product_id, quantity, amount):
        # 1. 验证参数
        if quantity <= 0 or amount <= 0:
            return {"code": 400, "message": "参数错误", "data": {}}
        
        # 2. 扣减库存
        product = self.product_repo.get(product_id)
        if not product or product["stock"] < quantity:
            return {"code": 400, "message": "库存不足", "data": {}}
        
        # 3. 创建订单
        order_id = f"{datetime.now().strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:6]}"
        order = {
            "order_id": order_id,
            "user_id": user_id,
            "product_id": product_id,
            "quantity": quantity,
            "amount": amount,
            "status": "pending",
            "create_time": datetime.now()
        }
        self.order_repo.create(order)
        
        # 4. 扣减库存
        self.product_repo.update_stock(product_id, product["stock"] - quantity)
        
        return {"code": 201, "message": "订单创建成功", "data": {"order_id": order_id}}
    
    def pay_order(self, order_id, channel):
        # 1. 验证订单
        order = self.order_repo.get(order_id)
        if not order or order["status"] != "pending":
            return {"code": 400, "message": "订单无效", "data": {}}
        
        # 2. 调用支付接口
        payment_result = self.payment_service.call_payment_api(order_id, order["amount"], channel)
        if payment_result["code"] != 200:
            return {"code": 400, "message": "支付失败", "data": {}}
        
        # 3. 更新订单状态
        order["status"] = "paid"
        order["pay_time"] = datetime.now()
        self.order_repo.update(order)
        
        # 4. 发送订单通知（异步）
        from app.services.notification_service import NotificationService
        notification_service = NotificationService()
        notification_service.send_order_paid_notification(order)
        
        return {"code": 200, "message": "支付成功", "data": {"order_id": order_id, "status": "paid"}}

# 乐观锁扣减库存
def update_stock(self, product_id, new_stock):
    # 更新条件包含当前库存，确保只有当库存未变化时才更新
    sql = "UPDATE product SET stock = %s WHERE id = %s AND stock = %s"
    result = self.db.execute(sql, (new_stock, product_id, product["stock"]))
    if result.rowcount == 0:
        raise StockUpdateError("库存更新失败，请重试")

# 用Celery异步发送通知
from app.tasks import send_order_paid_notification

# 在订单支付成功后调用
send_order_paid_notification.delay(order)

def handle_payment_callback(self, data):
    # 检查订单是否已经处理过
    order = self.order_repo.get(data["order_id"])
    if order["status"] == "paid":
        return {"code": 200, "message": "订单已处理", "data": {}}
    
    # 处理支付回调
    ...

from typing import Dict, Any
from app.exceptions import OrderNotFoundError

class OrderService:
    """订单服务类，处理订单的创建、支付、查询"""
    
    def get_order(self, order_id: str) -> Dict[str, Any]:
        """
        查询订单详情
        
        参数：
            order_id: 订单ID，格式为YYYYMMDDHHMMSS_XXXXXX
        返回：
            订单详情字典，包含订单ID、用户ID、商品ID、数量、金额、状态等
        抛出：
            OrderNotFoundError: 订单不存在时抛出
        """
        order = self.order_repo.get(order_id)
        if not order:
            raise OrderNotFoundError(f"订单{order_id}不存在")
        return order

name: CI

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.10
    - name: Install dependencies
      run: pip install -r requirements.txt
    - name: Run tests
      run: pytest
    - name: Check code style
      run: flake8
    - name: Check type hint
      run: mypy