软件开发概念配套练习

1
2
3
4

mkdir order-system && cd order-system
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install fastapi uvicorn sqlalchemy pydantic pytest httpx
git init

1
2
3
4
5

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
    return {"message": "Hello"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# main.py
from fastapi import FastAPI, Depends

app = FastAPI()

# ---------- 职责分离 ----------
class OrderValidator:
    def validate(self, order: dict):
        if order.get("total", 0) <= 0:
            raise ValueError("金额必须大于0")

class OrderRepository:
    def save(self, order: dict):
        print(f"订单 {order['id']} 已保存到数据库")

class EmailSender:
    def send(self, to: str, message: str):
        print(f"邮件发送至 {to}: {message}")

# ---------- 服务层（只做调度） ----------
class OrderService:
    def __init__(self, validator: OrderValidator,
                 repository: OrderRepository,
                 email_sender: EmailSender):
        self.validator = validator
        self.repository = repository
        self.email_sender = email_sender

    def process(self, order: dict):
        self.validator.validate(order)
        self.repository.save(order)
        self.email_sender.send(order["email"], "订单已创建")

# ---------- DI 容器（手动模拟） ----------
def get_order_service():
    return OrderService(OrderValidator(), OrderRepository(), EmailSender())

# ---------- 路由 ----------
@app.post("/orders")
def create_order(order: dict, service: OrderService = Depends(get_order_service)):
    service.process(order)
    return {"status": "ok"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from abc import ABC, abstractmethod

# ---------- 抽象接口 ----------
class PaymentMethod(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        pass

# ---------- 具体实现 ----------
class CreditCard(PaymentMethod):
    def pay(self, amount: float):
        return f"信用卡支付 {amount} 元"

class PayPal(PaymentMethod):
    def pay(self, amount: float):
        return f"PayPal 支付 {amount} 元"

# ---------- 使用方（对修改关闭） ----------
class Checkout:
    def __init__(self, payment: PaymentMethod):
        self.payment = payment

    def execute(self, amount: float):
        return self.payment.pay(amount)

# FastAPI 示例：通过查询参数选择支付方式
from fastapi import FastAPI, Query, Depends

app = FastAPI()

def get_payment(method: str = Query("credit")):
    if method == "credit":
        return CreditCard()
    elif method == "paypal":
        return PayPal()
    else:
        raise ValueError("不支持的支付方式")

@app.get("/pay")
def pay(amount: float, checkout: Checkout = Depends(lambda: Checkout(get_payment()))):
    # 实际项目中 Depends 里应根据请求参数构建 checkout
    return {"result": checkout.execute(amount)}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from abc import ABC, abstractmethod

# ---------- 抽象（高层模块依赖它） ----------
class Database(ABC):
    @abstractmethod
    def save(self, data: dict):
        pass

# ---------- 具体实现（低层模块） ----------
class PostgresDatabase(Database):
    def save(self, data: dict):
        print(f"Postgres 保存: {data}")

class InMemoryDatabase(Database):
    def __init__(self):
        self.storage = []

    def save(self, data: dict):
        self.storage.append(data)
        print("内存保存")

# ---------- 业务逻辑 ----------
class UserService:
    def __init__(self, db: Database):
        self.db = db

    def create_user(self, name: str):
        self.db.save({"name": name})

# FastAPI 使用依赖注入选择实现
from fastapi import FastAPI, Depends

app = FastAPI()

# 切换数据库只需改这里
def get_database() -> Database:
    return InMemoryDatabase()   # 或 PostgresDatabase()

@app.post("/users")
def create_user(name: str, service: UserService = Depends(lambda: UserService(get_database()))):
    service.create_user(name)
    return {"user": name}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 支付工厂
class PaymentFactory:
    @staticmethod
    def create(payment_type: str) -> PaymentMethod:
        if payment_type == "credit":
            return CreditCard()
        elif payment_type == "paypal":
            return PayPal()
        else:
            raise ValueError(f"未知支付方式: {payment_type}")

# 在 FastAPI 中使用
def get_payment_from_factory(method: str = "credit"):
    return PaymentFactory.create(method)

# 路由不变，工厂负责创建

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

from abc import ABC, abstractmethod

# ---------- 策略接口 ----------
class DiscountStrategy(ABC):
    @abstractmethod
    def apply(self, price: float) -> float:
        pass

class NoDiscount(DiscountStrategy):
    def apply(self, price):
        return price

class BlackFridayDiscount(DiscountStrategy):
    def apply(self, price):
        return price * 0.5

# ---------- 上下文 ----------
class PriceCalculator:
    def __init__(self, strategy: DiscountStrategy):
        self.strategy = strategy

    def calculate(self, price: float):
        return self.strategy.apply(price)

# FastAPI 依赖注入 + 工厂
from fastapi import FastAPI, Query

app = FastAPI()

def get_discount_strategy(campaign: str = Query("none")):
    if campaign == "blackfriday":
        return BlackFridayDiscount()
    return NoDiscount()

@app.get("/price")
def get_price(original: float, calculator: PriceCalculator = Depends(
    lambda campaign=Query("none"): PriceCalculator(get_discount_strategy(campaign))
)):
    # 实际中需要配合 Depends 传参，这里为了简洁直接 Lambda
    return {"final_price": calculator.calculate(original)}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

from typing import List

# ---------- 观察者接口 ----------
class OrderObserver:
    def on_payment_success(self, order_id: str, amount: float):
        pass

class InventoryObserver(OrderObserver):
    def on_payment_success(self, order_id, amount):
        print(f"[库存] 扣减库存，订单 {order_id}")

class PointsObserver(OrderObserver):
    def on_payment_success(self, order_id, amount):
        print(f"[积分] 增加积分，订单 {order_id}")

# ---------- 被观察者 ----------
class PaymentProcessor:
    def __init__(self, observers: List[OrderObserver] = None):
        self.observers = observers or []

    def add_observer(self, observer: OrderObserver):
        self.observers.append(observer)

    def pay(self, order_id: str, amount: float):
        print(f"支付成功，订单 {order_id}")
        for obs in self.observers:
            obs.on_payment_success(order_id, amount)

# FastAPI 组装
from fastapi import FastAPI, Depends

app = FastAPI()

def get_payment_processor():
    processor = PaymentProcessor()
    processor.add_observer(InventoryObserver())
    processor.add_observer(PointsObserver())
    return processor

@app.post("/pay/{order_id}")
def pay(order_id: str, amount: float,
        processor: PaymentProcessor = Depends(get_payment_processor)):
    processor.pay(order_id, amount)
    return {"status": "paid"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# test_price_calculator.py
import pytest
from main import PriceCalculator, NoDiscount, BlackFridayDiscount

def test_no_discount():
    calc = PriceCalculator(NoDiscount())
    assert calc.calculate(100) == 100

def test_black_friday():
    calc = PriceCalculator(BlackFridayDiscount())
    assert calc.calculate(100) == 50

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

from dataclasses import dataclass
from uuid import uuid4

# ---------- 值对象（不可变） ----------
@dataclass(frozen=True)
class Address:
    street: str
    city: str

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str = "CNY"

# ---------- 实体（有 ID，可变） ----------
class Order:
    def __init__(self, customer_name: str, address: Address):
        self.id = str(uuid4())
        self.customer_name = customer_name
        self.shipping_address = address
        self.items = []
        self.status = "NEW"

    def add_item(self, product: str, price: Money):
        self.items.append({"product": product, "price": price})

    def total(self) -> Money:
        total_amount = sum(item["price"].amount for item in self.items)
        return Money(total_amount)

# ---------- 仓储接口（领域层定义） ----------
class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order):
        pass

# ---------- 基础设施实现 ----------
class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._store = {}

    def save(self, order: Order):
        self._store[order.id] = order

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()

# ---------- 模型 ----------
class OrderCreate(BaseModel):
    customer_name: str
    amount: float

# ---------- 模拟数据库 ----------
orders_db = {}

# ---------- 路由 ----------
@app.post("/orders", status_code=201)
def create_order(order: OrderCreate):
    order_id = str(uuid4())
    orders_db[order_id] = order.dict()
    return {"id": order_id, **orders_db[order_id]}

@app.get("/orders/{order_id}")
def get_order(order_id: str):
    if order_id not in orders_db:
        raise HTTPException(status_code=404, detail="未找到")
    return orders_db[order_id]

@app.put("/orders/{order_id}")
def update_order(order_id: str, order: OrderCreate):
    if order_id not in orders_db:
        raise HTTPException(status_code=404, detail="未找到")
    orders_db[order_id] = order.dict()
    return {"id": order_id, **orders_db[order_id]}

@app.delete("/orders/{order_id}", status_code=204)
def delete_order(order_id: str):
    if order_id in orders_db:
        del orders_db[order_id]
    return

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import sqlite3
from fastapi import FastAPI, HTTPException

app = FastAPI()

def get_connection():
    conn = sqlite3.connect("bank.db")
    conn.execute("PRAGMA journal_mode=WAL")
    return conn

@app.post("/transfer")
def transfer(from_acc: str, to_acc: str, amount: float):
    conn = get_connection()
    try:
        conn.execute("BEGIN")
        # 检查余额
        cur = conn.execute("SELECT balance FROM accounts WHERE name=?", (from_acc,))
        row = cur.fetchone()
        if not row or row[0] < amount:
            raise ValueError("余额不足")
        # 扣减
        conn.execute("UPDATE accounts SET balance = balance - ? WHERE name=?", (amount, from_acc))
        # 增加
        conn.execute("UPDATE accounts SET balance = balance + ? WHERE name=?", (amount, to_acc))
        conn.commit()
        return {"status": "success"}
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        conn.close()