Python系统设计模式:分层架构与模块化

嘿，学习搭子，欢迎回来！上周咱们一起征服了异步编程与高并发设计，是不是感觉处理高负载场景更有底气了？这周咱们要深入另一个至关重要的领域——系统设计模式中的分层架构与模块化。这不仅是写代码的技巧，更是构建可维护、可扩展大型系统的核心思维。准备好了吗？咱们一起来搭建坚实的软件“摩天大楼”！

• • 纵向划分：按功能职责将系统划分为多个层次
• • 单向依赖：上层只能依赖下层，下层对上层透明
• • 接口通信：层间通过定义良好的接口交互，隐藏实现细节
• 

# 一个典型的未分层代码（所有逻辑混在一起）defhandle_user_request(request_data):# 1. 参数验证if'username'notin request_data or'password'notin request_data:return {'error': '参数缺失'}# 2. 业务逻辑：用户注册    username = request_data['username']    password = request_data['password']# 3. 直接数据库操作import sqlite3    conn = sqlite3.connect('users.db')    cursor = conn.cursor()# 检查用户是否存在    cursor.execute('SELECT * FROM users WHERE username = ?', (username,))if cursor.fetchone():return {'error': '用户已存在'}# 插入新用户    cursor.execute('INSERT INTO users (username, password) VALUES (?, ?)',                    (username, password))    conn.commit()# 4. 生成响应return {'success': True, 'message': '注册成功'}# 问题显而易见：# 1. 业务逻辑与数据访问紧耦合# 2. 难以单元测试# 3. 技术栈变更困难（如换数据库）# 4. 代码复用性差

1. 1. 职责清晰：每层只关注自己的事，降低理解成本
2. 2. 可维护性高：修改一层不影响其他层，减少“牵一发而动全身”
3. 3. 可测试性强：每层可独立测试，Mock依赖方便
4. 4. 技术栈隔离：底层技术变更不影响上层业务逻辑
5. 5. 团队协作高效：按层分工，并行开发

# 目录结构示例project/├── presentation/    # 表现层（Controller）│   ├── controllers/│   └── views/├── business/        # 业务逻辑层（Service）│   ├── services/│   └── models/└── data/           # 数据访问层（DAO/Repository）    ├── repositories/    └── entities/

• • 表现层（Presentation Layer）：处理用户交互，负责输入输出
• • 业务逻辑层（Business Logic Layer）：实现核心业务规则
• • 数据访问层（Data Access Layer）：负责数据持久化
• 

# 目录结构示例project/├── interfaces/      # 接口层（API/Web）├── application/     # 应用层（用例编排）├── domain/          # 领域层（业务核心）└── infrastructure/  # 基础设施层（技术实现）

• • 接口层（Interfaces）：对外暴露API，处理HTTP请求
• • 应用层（Application）：协调领域层完成业务流程
• • 领域层（Domain）：封装核心业务逻辑和规则
• • 基础设施层（Infrastructure）：提供数据库、缓存等技术实现

# 思想核心：一切外部依赖都是可插拔的适配器# 业务核心在中心，通过端口与外部世界交互

1. 1. 单向依赖原则
• • 上层可以依赖下层，下层不能依赖上层
• • 禁止反向依赖和循环依赖
2. 2. 接口隔离原则
• • 层间通过接口交互，隐藏实现细节
• • 接口应该小而专，避免“肥胖接口”
3. 3. 依赖倒置原则
• • 高层模块不依赖低层模块，二者都依赖抽象
• • 抽象不应该依赖细节，细节应该依赖抽象
4. 4. 开放封闭原则
• • 对扩展开放，对修改关闭
• • 通过抽象层隔离变化

• • 将系统分解为高内聚、低耦合的独立模块
• • 每个模块有明确的输入输出接口
• • 模块间通过接口通信，不直接依赖内部实现

# 基本模块示例# math_utils.py - 数学工具模块defadd(a, b):return a + bdefmultiply(a, b):return a * b# main.py - 主程序模块import math_utilsresult = math_utils.add(5, 3)print(f"5 + 3 = {result}")

# 复杂的项目结构my_project/├── src/│   ├── core/               # 核心模块│   │   ├── __init__.py│   │   ├── models.py      # 领域模型│   │   └── services.py    # 核心服务│   ├── infrastructure/    # 基础设施模块│   │   ├── __init__.py│   │   ├── database.py    # 数据库操作│   │   └── cache.py       # 缓存管理│   └── interfaces/        # 接口模块│       ├── __init__.py│       ├── web_api.py     # Web API│       └── cli.py         # 命令行接口└── tests/                 # 测试模块    ├── test_models.py    └── test_services.py

# 不好的做法：所有功能在一个文件# user_manager.py 包含：认证、授权、资料管理、通知...# 好的做法：按功能拆分# authentication.py - 认证逻辑# authorization.py  - 权限管理# profile.py       - 资料管理# notification.py  - 通知服务

# 定义接口模块 interfaces/user_service.pyfrom abc import ABC, abstractmethodfrom typing importOptionalfrom ..domain.models import UserclassIUserService(ABC):"""用户服务接口"""    @abstractmethoddefregister(self, username: str, password: str) -> User:"""注册新用户"""pass    @abstractmethoddefauthenticate(self, username: str, password: str) -> Optional[User]:"""用户认证"""pass    @abstractmethoddefupdate_profile(self, user_id: int, profile_data: dict) -> User:"""更新用户资料"""pass

# 紧耦合的代码classOrderService:def__init__(self):self.db = MySQLDatabase()  # 直接依赖具体实现defcreate_order(self, order_data):# 使用self.db操作数据库pass# 解耦后的代码classOrderService:def__init__(self, database):  # 依赖抽象self.db = databasedefcreate_order(self, order_data):# 使用self.db操作数据库pass# 使用db = MySQLDatabase()  # 或 PostgreSQLDatabase()service = OrderService(db)

# 定义仓储接口from abc import ABC, abstractmethodfrom typing importList, Optionalfrom domain.models import UserclassIUserRepository(ABC):    @abstractmethoddefget_by_id(self, user_id: int) -> Optional[User]:pass    @abstractmethoddefget_by_username(self, username: str) -> Optional[User]:pass    @abstractmethoddefsave(self, user: User) -> User:pass    @abstractmethoddefdelete(self, user_id: int) -> bool:pass# 具体实现classUserRepository(IUserRepository):def__init__(self, session):self.session = sessiondefget_by_id(self, user_id: int) -> Optional[User]:returnself.session.query(User).filter(User.id == user_id).first()# 其他方法实现...

# 业务服务层classUserService:def__init__(self, user_repository: IUserRepository, email_service):self.user_repository = user_repositoryself.email_service = email_servicedefregister_user(self, username: str, password: str, email: str):# 业务逻辑ifself.user_repository.get_by_username(username):raise ValueError("用户名已存在")# 创建用户        user = User(username=username, password=password, email=email)        saved_user = self.user_repository.save(user)# 发送欢迎邮件self.email_service.send_welcome_email(email)return saved_user

classUnitOfWork:def__init__(self):self.users = UserRepository()self.orders = OrderRepository()self._is_committed = Falsedefcommit(self):# 提交所有更改self._is_committed = Truedefrollback(self):# 回滚所有更改self._is_committed = Falsedef__enter__(self):returnselfdef__exit__(self, exc_type, exc_val, exc_tb):if exc_type:self.rollback()else:self.commit()

# 传统方式：紧耦合classUserService:def__init__(self):self.db = MySQLDatabase()  # 直接创建依赖defget_user(self, user_id):returnself.db.query(f"SELECT * FROM users WHERE id = {user_id}")# 依赖注入：松耦合classUserService:def__init__(self, database):  # 依赖由外部传入self.db = databasedefget_user(self, user_id):returnself.db.query(f"SELECT * FROM users WHERE id = {user_id}")# 使用db = MySQLDatabase()  # 或 PostgreSQLDatabase()service = UserService(db)  # 依赖被注入

1. 1. 依赖查找（Dependency Lookup）：主动从容器中查找依赖
2. 2. 依赖注入（Dependency Injection）：被动接收容器注入的依赖

classOrderService:def__init__(self, order_repository, payment_gateway, email_service):self.order_repository = order_repositoryself.payment_gateway = payment_gatewayself.email_service = email_servicedefcreate_order(self, order_data):# 使用注入的依赖        order = self.order_repository.save(order_data)self.payment_gateway.charge(order.total_amount)self.email_service.send_order_confirmation(order.email)return order

classUserService:def__init__(self):self.user_repository = None# 初始为空self.logger = Nonedefset_user_repository(self, repository):self.user_repository = repositorydefset_logger(self, logger):self.logger = loggerdefget_user(self, user_id):ifself.logger:self.logger.info(f"查询用户: {user_id}")returnself.user_repository.get_by_id(user_id)

from abc import ABC, abstractmethodclassInjectable(ABC):    @abstractmethoddefinject(self, dependency):passclassUserService(Injectable):def__init__(self):self.db = Nonedefinject(self, database):self.db = database

# 创建所有依赖db = MySQLDatabase(host='localhost', user='root', password='secret')logger = FileLogger('app.log')email_sender = SMTPEmailSender()# 手动注入user_repository = UserRepository(db)user_service = UserService(user_repository, logger, email_sender)# 使用user = user_service.get_user(123)

from dependency_injector import containers, providersfrom dependency_injector.wiring import inject, Provide# 定义服务类classDatabase:defquery(self, sql):print(f"执行查询: {sql}")return {"id": 1, "name": "张三"}classUserService:    @injectdef__init__(self, db: Database = Provide['db']):self.db = dbdefget_user(self, user_id):returnself.db.query(f"SELECT * FROM users WHERE id = {user_id}")# 定义容器classContainer(containers.DeclarativeContainer):    db = providers.Singleton(Database)    user_service = providers.Factory(UserService, db=db)# 使用容器container = Container()service = container.user_service()user = service.get_user(123)print(f"获取用户: {user}")

from abc import ABC, abstractmethodfrom typing importDict, Any# 定义数据库接口classIDatabase(ABC):    @abstractmethoddefconnect(self) -> None:pass    @abstractmethoddefdisconnect(self) -> None:pass    @abstractmethoddefexecute(self, sql: str, params: Dict = None) -> Any:pass# MySQL实现classMySQLDatabase(IDatabase):def__init__(self, host, port, user, password, database):self.host = hostself.port = portself.user = userself.password = passwordself.database = databaseself.connection = Nonedefconnect(self):import mysql.connectorself.connection = mysql.connector.connect(            host=self.host,            port=self.port,            user=self.user,            password=self.password,            database=self.database        )defdisconnect(self):ifself.connection:self.connection.close()defexecute(self, sql: str, params: Dict = None):        cursor = self.connection.cursor()        cursor.execute(sql, params or {})return cursor.fetchall()# PostgreSQL实现classPostgreSQLDatabase(IDatabase):def__init__(self, connection_string):self.connection_string = connection_stringself.connection = Nonedefconnect(self):import psycopg2self.connection = psycopg2.connect(self.connection_string)# 其他方法类似...

classConfig:def__init__(self, config_file='config.json'):self.config_file = config_fileself._config = self._load_config()def_load_config(self):import jsonwithopen(self.config_file, 'r', encoding='utf-8') as f:return json.load(f)defget(self, key, default=None):returnself._config.get(key, default)# 服务类通过依赖注入获取配置classAppService:def__init__(self, config: Config):self.config = configself.api_key = config.get('api_key')self.base_url = config.get('base_url')defmake_request(self, endpoint):import requests        url = f"{self.base_url}/{endpoint}"        headers = {"Authorization": f"Bearer {self.api_key}"}return requests.get(url, headers=headers)

1. 1. 用户注册与登录
2. 2. 商品浏览与搜索
3. 3. 购物车管理
4. 4. 订单创建与支付
5. 5. 订单状态查询

# 完整的项目目录结构ecommerce_project/├── src/│   ├── domain/                     # 领域层（业务核心）│   │   ├── entities/              # 实体│   │   │   ├── user.py│   │   │   ├── product.py│   │   │   ├── order.py│   │   │   └── cart.py│   │   ├── value_objects/         # 值对象│   │   │   ├── money.py│   │   │   └── address.py│   │   ├── services/              # 领域服务│   │   │   ├── order_service.py│   │   │   └── payment_service.py│   │   └── events/                # 领域事件│   │       ├── order_created.py│   │       └── payment_completed.py│   ││   ├── application/                # 应用层（用例编排）│   │   ├── usecases/│   │   │   ├── create_order_usecase.py│   │   │   └── process_payment_usecase.py│   │   └── dto/                    # 数据传输对象│   │       ├── order_dto.py│   │       └── user_dto.py│   ││   ├── infrastructure/             # 基础设施层│   │   ├── database/│   │   │   ├── repositories/│   │   │   ├── migrations/│   │   │   └── session.py│   │   ├── cache/│   │   ├── external/│   │   │   ├── payment_gateway.py│   │   │   └── email_service.py│   │   └── config/                 # 配置管理│   │       └── settings.py│   ││   └── interfaces/                 # 接口层│       ├── web/│       │   ├── controllers/│       │   ├── middlewares/│       │   └── routers.py│       ├── cli/                    # 命令行接口│       └── api/                    # API接口定义│├── tests/                          # 测试目录│   ├── unit/│   ├── integration/│   └── e2e/│├── requirements.txt                # 依赖管理├── Dockerfile                      # 容器化├── docker-compose.yml└── README.md

# src/domain/entities/order.pyfrom dataclasses import dataclassfrom datetime import datetimefrom typing importListfrom ..value_objects.money import Moneyfrom ..events.order_created import OrderCreated@dataclassclassOrderItem:    product_id: int    product_name: str    unit_price: Money    quantity: int    @propertydeftotal_price(self) -> Money:return Money(self.unit_price.amount * self.quantity, self.unit_price.currency)@dataclassclassOrder:id: int    user_id: int    items: List[OrderItem]    status: str# 'pending', 'paid', 'shipped', 'delivered', 'cancelled'    created_at: datetime    updated_at: datetime    _events: List = Nonedef__post_init__(self):self._events = []    @propertydeftotal_amount(self) -> Money:        total = Money(0, 'CNY') ifself.items else Money(0, 'CNY')for item inself.items:            total = total.add(item.total_price)return totaldefmark_as_paid(self):ifself.status != 'pending':raise ValueError(f"订单状态为{self.status}，不能标记为已支付")self.status = 'paid'self.updated_at = datetime.now()defcancel(self):ifself.status in ['shipped', 'delivered']:raise ValueError(f"订单状态为{self.status}，不能取消")self.status = 'cancelled'self.updated_at = datetime.now()defget_events(self):        events = self._events.copy()self._events.clear()return eventsdef_add_event(self, event):self._events.append(event)

# src/domain/services/order_service.pyfrom typing importOptionalfrom ..entities.order import Orderfrom ..events.order_created import OrderCreatedfrom ...infrastructure.database.repositories.order_repository import IOrderRepositoryclassOrderService:def__init__(self, order_repository: IOrderRepository):self.order_repository = order_repositorydefcreate_order(self, user_id: int, items_data: list) -> Order:# 验证业务规则ifnot items_data:raise ValueError("订单必须包含至少一件商品")# 创建订单项        order_items = []for item_data in items_data:            item = OrderItem(                product_id=item_data['product_id'],                product_name=item_data['product_name'],                unit_price=Money(item_data['unit_price'], 'CNY'),                quantity=item_data['quantity']            )            order_items.append(item)# 创建订单        order = Order(id=None,  # 由数据库生成            user_id=user_id,            items=order_items,            status='pending',            created_at=datetime.now(),            updated_at=datetime.now()        )# 保存订单        saved_order = self.order_repository.save(order)# 记录领域事件        order._add_event(OrderCreated(            order_id=saved_order.id,            user_id=user_id,            total_amount=saved_order.total_amount.amount,            items_count=len(items_data)        ))return saved_orderdefget_order(self, order_id: int, user_id: Optional[int] = None) -> Optional[Order]:        order = self.order_repository.get_by_id(order_id)# 权限检查：用户只能查看自己的订单（除非是管理员）if user_id and order and order.user_id != user_id:raise PermissionError("无权查看此订单")return order

# src/application/usecases/create_order_usecase.pyfrom typing importDict, Anyfrom ...domain.services.order_service import OrderServicefrom ...infrastructure.external.payment_gateway import IPaymentGatewayfrom ...infrastructure.external.email_service import IEmailServicefrom ..dto.order_dto import OrderDTOclassCreateOrderUseCase:def__init__(        self,        order_service: OrderService,        payment_gateway: IPaymentGateway,        email_service: IEmailService):self.order_service = order_serviceself.payment_gateway = payment_gatewayself.email_service = email_servicedefexecute(self, user_id: int, items_data: list) -> Dict[str, Any]:# 1. 创建订单        order = self.order_service.create_order(user_id, items_data)# 2. 调用支付网关        payment_result = self.payment_gateway.create_payment(            amount=order.total_amount.amount,            currency=order.total_amount.currency,            order_id=order.id        )# 3. 发送订单确认邮件self.email_service.send_order_confirmation(            user_id=user_id,            order_id=order.id,            total_amount=order.total_amount.amount        )# 4. 返回DTOreturn OrderDTO.from_entity(order).to_dict()

# src/interfaces/web/controllers/order_controller.pyfrom typing importDict, Anyfrom fastapi import APIRouter, Depends, HTTPExceptionfrom ...application.usecases.create_order_usecase import CreateOrderUseCasefrom ...application.usecases.get_order_usecase import GetOrderUseCasefrom ...infrastructure.config.settings import get_settingsfrom ...infrastructure.database.session import get_dbfrom .dependencies import get_current_userrouter = APIRouter(prefix="/orders", tags=["orders"])@router.post("/", response_model=Dict[str, Any])asyncdefcreate_order(    order_data: Dict[str, Any],    current_user: Dict = Depends(get_current_user),    usecase: CreateOrderUseCase = Depends()):try:        user_id = current_user['id']        result = usecase.execute(user_id, order_data['items'])return {"success": True, "data": result}except ValueError as e:raise HTTPException(status_code=400, detail=str(e))except Exception as e:# 记录日志raise HTTPException(status_code=500, detail="内部服务器错误")@router.get("/{order_id}", response_model=Dict[str, Any])asyncdefget_order(    order_id: int,    current_user: Dict = Depends(get_current_user),    usecase: GetOrderUseCase = Depends()):try:        user_id = current_user['id']        order = usecase.execute(order_id, user_id)return {"success": True, "data": order}except PermissionError as e:raise HTTPException(status_code=403, detail=str(e))except Exception as e:raise HTTPException(status_code=500, detail="内部服务器错误")

# src/infrastructure/container.pyfrom dependency_injector import containers, providersfrom .database.session import SessionLocalfrom .database.repositories.order_repository import OrderRepositoryfrom .database.repositories.user_repository import UserRepositoryfrom .external.payment_gateway import StripePaymentGatewayfrom .external.email_service import SendGridEmailServicefrom ..domain.services.order_service import OrderServicefrom ..domain.services.user_service import UserServicefrom ..application.usecases.create_order_usecase import CreateOrderUseCasefrom ..application.usecases.get_order_usecase import GetOrderUseCaseclassContainer(containers.DeclarativeContainer):# 配置    config = providers.Configuration()# 数据库会话    db_session = providers.Singleton(SessionLocal)# 仓储    order_repository = providers.Factory(        OrderRepository,        session=db_session    )    user_repository = providers.Factory(        UserRepository,        session=db_session    )# 外部服务    payment_gateway = providers.Factory(        StripePaymentGateway,        api_key=config.stripe.api_key    )    email_service = providers.Factory(        SendGridEmailService,        api_key=config.sendgrid.api_key    )# 领域服务    order_service = providers.Factory(        OrderService,        order_repository=order_repository    )    user_service = providers.Factory(        UserService,        user_repository=user_repository    )# 应用层用例    create_order_usecase = providers.Factory(        CreateOrderUseCase,        order_service=order_service,        payment_gateway=payment_gateway,        email_service=email_service    )    get_order_usecase = providers.Factory(        GetOrderUseCase,        order_service=order_service    )# 全局容器实例container = Container()

# 简化的分层架构示例from datetime import datetimefrom typing importList, Optional, Dict, Anyfrom dataclasses import dataclass, field# ---------- 值对象 ----------@dataclassclassMoney:    amount: float    currency: strdefadd(self, other: 'Money') -> 'Money':ifself.currency != other.currency:raise ValueError("货币单位不一致")return Money(self.amount + other.amount, self.currency)# ---------- 领域实体 ----------@dataclassclassOrderItem:    product_id: int    product_name: str    unit_price: Money    quantity: int    @propertydeftotal_price(self) -> Money:return Money(self.unit_price.amount * self.quantity, self.unit_price.currency)@dataclassclassOrder:id: Optional[int]    user_id: int    items: List[OrderItem]    status: str    created_at: datetime    updated_at: datetime    _events: List = field(default_factory=list)# ---------- 仓储接口 ----------classIOrderRepository:defsave(self, order: Order) -> Order:passdefget_by_id(self, order_id: int) -> Optional[Order]:pass# ---------- 领域服务 ----------classOrderService:def__init__(self, repository: IOrderRepository):self.repository = repositorydefcreate_order(self, user_id: int, items: List[Dict]) -> Order:# 业务逻辑ifnot items:raise ValueError("订单必须包含商品")# 创建订单        order = Order(id=None,            user_id=user_id,            items=[],            status='pending',            created_at=datetime.now(),            updated_at=datetime.now()        )# 添加订单项for item in items:            order_item = OrderItem(                product_id=item['product_id'],                product_name=item['product_name'],                unit_price=Money(item['price'], 'CNY'),                quantity=item['quantity']            )            order.items.append(order_item)# 保存        order.id = 1001# 模拟数据库生成IDreturn order# ---------- 应用层 ----------classCreateOrderUseCase:def__init__(self, order_service: OrderService):self.order_service = order_servicedefexecute(self, user_id: int, items: List[Dict]) -> Dict[str, Any]:        order = self.order_service.create_order(user_id, items)return {'order_id': order.id,'user_id': order.user_id,'status': order.status,'total_amount': sum(item.total_price.amount for item in order.items),'items_count': len(order.items)        }# ---------- 主程序 ----------if __name__ == "__main__":# 1. 创建仓储（这里用内存实现）classMemoryOrderRepository(IOrderRepository):def__init__(self):self.orders = {}defsave(self, order: Order) -> Order:            order.id = len(self.orders) + 1000self.orders[order.id] = orderreturn orderdefget_by_id(self, order_id: int) -> Optional[Order]:returnself.orders.get(order_id)# 2. 创建依赖    repository = MemoryOrderRepository()    order_service = OrderService(repository)    usecase = CreateOrderUseCase(order_service)# 3. 执行用例    items = [        {"product_id": 1, "product_name": "Python编程书", "price": 99.0, "quantity": 2},        {"product_id": 2, "product_name": "Coffee", "price": 25.0, "quantity": 1}    ]    result = usecase.execute(user_id=123, items=items)# 4. 输出结果print("🎉 订单创建成功！")print(f"订单号: {result['order_id']}")print(f"用户ID: {result['user_id']}")print(f"订单状态: {result['status']}")print(f"商品数量: {result['items_count']}")print(f"总金额: ¥{result['total_amount']:.2f}")

# 过度分层的代码结构project/├── presentation/│   ├── controllers/│   ├── view_models/│   ├── presenters/│   └── formatters/├── application/│   ├── usecases/│   ├── services/│   ├── handlers/│   └── validators/├── domain/│   ├── models/│   ├── services/│   ├── factories/│   └── specifications/├── infrastructure/│   ├── repositories/│   ├── gateways/│   ├── adapters/│   └── clients/└── shared/    ├── utils/    ├── helpers/    ├── constants/    └── exceptions/# 一个简单的查询需要经过7层调用：# Controller → UseCase → DomainService → Repository → Database# 然后原路返回：Database → Repository → DomainService → UseCase → Controller

• • 简单项目从三层开始：表现层、业务层、数据层
• • 按需增加层次：当系统复杂度确实需要时才增加
• • 避免"为了分层而分层"：每层都应有明确的、不可替代的职责

# 贫血模型：实体只有数据，没有行为@dataclassclassOrder:id: int    items: List[OrderItem]    status: str    total_amount: float# 只有数据字段，没有任何业务方法# 业务逻辑散落在Service中classOrderService:defmark_order_as_paid(self, order: Order):if order.status != 'pending':raise ValueError("订单状态错误")        order.status = 'paid'# 这里应该属于Order实体的业务逻辑defcalculate_total(self, order: Order):        total = 0for item in order.items:            total += item.price * item.quantity        order.total_amount = total# 这里应该属于Order实体的计算逻辑

• • 丰富领域模型：将与实体状态相关的逻辑放在实体内部
• • 遵循"信息专家模式"：谁拥有数据，谁就负责操作
• • 适度充血：不是所有逻辑都要放在实体中，只有与实体状态强相关的逻辑才应该内聚

# 循环依赖：A依赖B，B又依赖A# domain/models/user.pyfrom ..services.order_service import OrderService  # 反向依赖！classUser:defget_recent_orders(self):        service = OrderService()  # 错误：领域层依赖应用层return service.get_orders_by_user(self.id)# application/services/order_service.pyfrom ...domain.models.user import User  # 循环依赖classOrderService:defget_orders_by_user(self, user_id):        user = User.get_by_id(user_id)  # 错误：应用层依赖领域模型# ...

• • 严格单向依赖：上层可以依赖下层，下层不能依赖上层
• • 依赖倒置：通过接口抽象解耦
• • 依赖注入：避免直接创建依赖对象

# 典型的N+1查询问题classOrderService:defget_user_orders_with_details(self, user_id):        orders = self.order_repository.get_by_user_id(user_id)  # 1次查询        result = []for order in orders:# 为每个订单单独查询用户信息            user = self.user_repository.get_by_id(order.user_id)  # N次查询            order.user_name = user.name            result.append(order)return result  # 总共1+N次查询

• • 批量查询：使用IN语句或JOIN查询
• • 缓存策略：合理使用缓存减少数据库访问
• • 延迟加载：只有在需要时才加载关联数据

# 业务层直接依赖具体技术实现classProductService:def__init__(self):# 直接依赖MySQL连接self.conn = mysql.connector.connect(            host='localhost',            user='root',            password='secret',            database='products'        )defget_product(self, product_id):        cursor = self.conn.cursor()        cursor.execute('SELECT * FROM products WHERE id = %s', (product_id,))# ...

• • 定义接口抽象：业务层只依赖接口，不依赖实现
• • 依赖注入：通过容器管理技术实现
• • 适配器模式：为不同技术实现提供统一接口

1. 1. 保持简单：从三层架构开始，按需扩展
2. 2. 明确边界：每层有明确的输入输出接口
3. 3. 依赖抽象：通过接口隔离具体实现
4. 4. 合理分层：避免过度设计，保持每层都有实际价值
5. 5. 性能意识：注意层间调用的性能开销
6. 6. 持续重构：随着业务发展调整架构

1. 1. 基础理解：三层架构与四层架构（包含领域层）的主要区别是什么？在什么情况下应该选择四层架构？
2. 2. 应用分析：假设你要重构一个已有的单体应用，该应用目前所有代码都混在一起。请设计一个逐步重构为分层架构的方案，并说明每一步的关键决策。
3. 3. 深度思考：依赖注入虽然能解耦，但也会增加代码复杂度。在什么规模的项目中应该引入依赖注入框架？什么情况下手动依赖注入就足够了？
4. 4. 实践探索：使用分层架构实现一个简单的博客系统，要求：
• • 支持用户注册、登录
• • 支持文章创建、编辑、删除
• • 支持文章列表展示、详情查看
• • 使用接口抽象数据库访问层
• • 编写单元测试覆盖核心业务逻辑
5. 5. 综合挑战：现有系统使用传统三层架构，但随着业务发展，业务逻辑变得越来越复杂。你会如何将其演进到领域驱动设计（DDD）的分层架构？需要考虑哪些迁移风险和成本？

1. 1. 架构模式深入：
• • 学习六边形架构（端口与适配器）
• • 研究清洁架构（Clean Architecture）
• • 探索CQRS（命令查询职责分离）模式
2. 2. 实践框架学习：
• • Spring Boot的分层实践
• • Django的MTV模式与分层
• • FastAPI的依赖注入系统
3. 3. 相关技术栈：
• • 数据库迁移工具（Alembic、Flyway）
• • API文档生成（OpenAPI、Swagger）
• • 容器化部署（Docker、Kubernetes）