第 1 章：后端架构概述¶

# 单体架构示例
class MonolithicApp:
    def __init__(self):
        self.user_service = UserService()
        self.order_service = OrderService()
        self.payment_service = PaymentService()

    def create_order(self, user_id, product_id):
        # 所有服务在一个应用中
        user = self.user_service.get_user(user_id)
        order = self.order_service.create_order(user, product_id)
        payment = self.payment_service.process_payment(order)
        return order

# 垂直拆分架构示例
# 用户服务
class UserService:
    def get_user(self, user_id):
        pass

# 订单服务
class OrderService:
    def create_order(self, user_id, product_id):
        pass

# 支付服务
class PaymentService:
    def process_payment(self, order_id):
        pass

# 微服务架构示例
from fastapi import FastAPI
import requests

# 用户服务
user_app = FastAPI()

@user_app.get("/users/{user_id}")
async def get_user(user_id: int):  # async def定义异步函数；用await调用
    return {"user_id": user_id, "name": "John"}

# 订单服务
order_app = FastAPI()

@order_app.post("/orders")
async def create_order(user_id: int, product_id: int):
    # 调用用户服务
    user = requests.get(f"http://user-service/users/{user_id}").json()
    # 创建订单
    return {"order_id": 1, "user": user}

# 违反SRP
class UserService:
    def create_user(self, user):
        pass

    def send_email(self, email):
        pass  # 发送邮件不是用户服务的职责

# 符合SRP
class UserService:
    def create_user(self, user):
        pass

class EmailService:
    def send_email(self, email):
        pass

# 违反OCP
class PaymentProcessor:
    def process(self, payment_type):
        if payment_type == "alipay":
            # 支付宝逻辑
            pass
        elif payment_type == "wechat":
            # 微信逻辑
            pass

# 符合OCP
class PaymentProcessor:
    def process(self, payment):
        payment.pay()

class AlipayPayment:
    def pay(self):
        pass

class WechatPayment:
    def pay(self):
        pass

# 违反LSP：矩形和正方形的经典问题
class Rectangle:
    def set_width(self, w): self._w = w
    def set_height(self, h): self._h = h
    def area(self): return self._w * self._h

class Square(Rectangle):
    def set_width(self, w):
        self._w = self._h = w  # 违反LSP：父类行为被改变
    def set_height(self, h):
        self._w = self._h = h

def process_shape(rect: Rectangle):
    rect.set_width(5)
    rect.set_height(10)
    assert rect.area() == 50  # Square 会失败！

# 符合LSP：使用抽象基类
from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self) -> float: ...

class Rectangle(Shape):
    def __init__(self, w, h): self._w, self._h = w, h
    def area(self): return self._w * self._h

class Square(Shape):
    def __init__(self, s): self._s = s
    def area(self): return self._s * self._s

# 违反ISP：一个大接口强迫实现不需要的方法
from abc import ABC, abstractmethod

class Worker(ABC):
    @abstractmethod
    def work(self): ...
    @abstractmethod
    def eat(self): ...   # 机器人不需要吃东西

class Robot(Worker):
    def work(self): print("robot working")
    def eat(self): raise NotImplementedError  # 被迫实现无意义的方法

# 符合ISP：拆分为独立接口
class Workable(ABC):
    @abstractmethod
    def work(self): ...

class Eatable(ABC):
    @abstractmethod
    def eat(self): ...

class Human(Workable, Eatable):
    def work(self): print("human working")
    def eat(self): print("human eating")

class Robot(Workable):
    def work(self): print("robot working")
    # 无需实现 eat()

# 违反DIP：高层模块直接依赖具体低层模块
class MySQLDatabase:
    def save(self, data): print(f"MySQL: saving {data}")

class UserService:
    def __init__(self):
        self.db = MySQLDatabase()  # 紧耦合，难以切换数据库

# 符合DIP：依赖抽象接口
from abc import ABC, abstractmethod

class Database(ABC):
    @abstractmethod
    def save(self, data: dict): ...

class MySQLDatabase(Database):
    def save(self, data): print(f"MySQL: saving {data}")

class MongoDatabase(Database):
    def save(self, data): print(f"MongoDB: saving {data}")

class UserService:
    def __init__(self, db: Database):  # 依赖抽象，可注入任意实现
        self.db = db

# 使用
service = UserService(MySQLDatabase())
service = UserService(MongoDatabase())  # 轻松切换

# 违反DRY
def get_user_name(user_id):
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    return user.name

def get_user_email(user_id):
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    return user.email

# 符合DRY
def get_user(user_id):
    return db.query("SELECT * FROM users WHERE id = ?", user_id)

def get_user_name(user_id):
    return get_user(user_id).name

def get_user_email(user_id):
    return get_user(user_id).email

# 违反KISS：过度设计的字符串拼接
class StringBuilderFactoryManagerImpl:
    def create_builder(self):
        return StringBuilder()

class StringBuilder:
    def __init__(self):
        self._parts = []
    def append(self, s):
        self._parts.append(s)
        return self
    def build(self):
        return "".join(self._parts)

factory = StringBuilderFactoryManagerImpl()
result = factory.create_builder().append("hello").append(" ").append("world").build()

# 符合KISS：直接用内置方法即可
result = "hello" + " " + "world"
# 或更 Pythonic
result = " ".join(["hello", "world"])

# 违反YAGNI：为"可能的未来需求"提前实现
class UserService:
    def create_user(self, name: str, email: str):
        # 当前只需要 name 和 email
        pass

    # 下面这些方法现在根本用不到
    def export_to_csv(self): ...       # 也许将来要导出？
    def sync_to_ldap(self): ...        # 也许将来要 LDAP？
    def generate_qr_code(self): ...    # 也许将来要二维码？

# 符合YAGNI：只实现当前需要的
class UserService:
    def create_user(self, name: str, email: str):
        # 简洁、专注，等真正需要时再扩展
        return {"name": name, "email": email}