你的域名不了解 PostgreSQL(而且不应该)

发布: (2026年3月25日 GMT+8 19:32)
5 分钟阅读
原文: Dev.to

Source: Dev.to

请提供您希望翻译的正文内容(除代码块、URL 和上述来源链接外),我将把它翻译成简体中文并保持原有的 Markdown 格式。

问题:业务逻辑耦合到基础设施

你的业务逻辑不应该关心你使用的是 PostgreSQL、MySQL,还是一堆文本文件。如果它关心,这不仅仅是小小的代码味道——而是架构问题。

纠结的示例

from sqlalchemy.orm import Session
from app.models import Order, Customer
from app.db import get_db
from fastapi import HTTPException
import os

class OrderService:
    def place_order(self, customer_id: int, items: list, db: Session):
        customer = db.query(Customer).filter(Customer.id == customer_id).first()
        if not customer:
            raise HTTPException(status_code=404, detail="Customer not found")

        total = sum(item["price"] * item["qty"] for item in items)

        if customer.loyalty_points > 500:
            total *= 0.9  # 10% discount

        order = Order(
            customer_id=customer_id,
            total=total,
            status="pending",
            items=items
        )
        db.add(order)
        db.commit()
        db.refresh(order)

        if os.getenv("NOTIFY_SERVICE_ENABLED") == "true":
            # fire and forget
            pass

        return {"order_id": order.id, "total": total, "status": "pending"}

这个函数了解以下内容:

  • SQLAlchemy 会话
  • FastAPI HTTP 响应和状态码
  • ORM 模型内部细节
  • 环境变量
  • 原始 JSON 字典结构

唯一的业务职责是决定订单是否有效以及如何定价。但它同时做了五件事,导致难以测试。

def test_loyalty_discount_applied():
    service = OrderService()
    # ...now what?
    # We need a real db Session
    # We need real Customer rows in that database
    # We need FastAPI's exception handling wired up
    # We need environment variables set correctly

常见结果:

  1. 启动一个测试数据库并编写冗长的 fixture,只为测试几行折扣逻辑。
  2. 过度 Mock SQLAlchemy,导致测试不再验证真实行为。
  3. 直接跳过测试(大多数团队的做法)。

最后一个选项并非懒惰;它是代码难以测试的症状。

提取纯粹的业务规则

def calculate_order_total(items: list, loyalty_points: int) -> float:
    total = sum(item["price"] * item["qty"] for item in items)
    if loyalty_points > 500:
        total *= 0.9
    return total

这才是业务真正关心的领域逻辑。所有外围的数据库调用、HTTP 处理和环境检查都是基础设施关注点,应该对计算保持透明。

一个干净的领域模型

from dataclasses import dataclass
from typing import List

@dataclass
class OrderItem:
    product_id: str
    price: float
    quantity: int

@dataclass
class Order:
    customer_id: str
    items: List[OrderItem]
    loyalty_points: int

    def calculate_total(self) -> float:
        total = sum(item.price * item.quantity for item in self.items)
        if self.loyalty_points > 500:
            total *= 0.9
        return total

    def is_valid(self) -> bool:
        return len(self.items) > 0 and all(item.quantity > 0 for item in self.items)

没有 SQLAlchemy、FastAPI、os.getenv 或原始字典——只有纯粹的领域概念。

测试领域逻辑

def test_loyalty_discount():
    order = Order(
        customer_id="c-1",
        items=[OrderItem("p-1", 100.0, 2)],
        loyalty_points=600
    )
    assert order.calculate_total() == 180.0  # 200 * 0.9

三行代码,零 fixture,零 mock,零数据库。测试在毫秒级完成。

为什么分离很重要

大多数后端代码库会把两件事混为一谈:

  1. 软件的功能 – 定义产品的规则和计算。
  2. 软件的实现方式 – 数据库、HTTP 框架、队列等。

当这两件事纠缠在一起时:

  • 你无法在不触及业务规则的情况下更换 PostgreSQL。
  • 你无法在不启动数据库的情况下测试业务规则。
  • 你无法在不牵连整个基础设施层的情况下在服务之间共享逻辑。

你的领域模型不应该了解 PostgreSQL(或任何特定的存储)。如果了解了,你就做出了隐式的架构决定,随着时间推移会付出代价。

保持分离

端口与适配器(又称六边形架构)提供了明确命名且始终如一强制执行的正式边界。它们保持领域模型的纯粹,同时允许外层独立地进行更改。

可操作步骤: 打开代码库中的任意服务类(例如 OrderServiceUserService)。统计它导入了多少与业务规则毫无关联的外部依赖。如果数量大于零,则说明你正面临本文所描述的问题。你并不孤单,而且有一条清晰的解决之路。

0 浏览
Back to Blog

相关文章

阅读更多 »