# ❌ 糟糕：机械、重复且容易手抖写错变量名
profile.bio = data.get('bio', profile.bio)
profile.location = data.get('location', profile.location)
profile.website = data.get('website', profile.website)
profile.phone = data.get('phone', profile.phone)
# ... 重复 10 次 ...
profile.save()

# ❌ 风险：逻辑复杂时极易漏掉保存操作
def update_user_status(user, data):
    if data.get('is_active'):
        user.is_active = True
        # 这里忘了写 user.save()，导致 Bug 极难排查
    ...

# ❌ 痛苦：为了判断变化，你得手动记录旧值，代码极其啰嗦
old_email = user.email
user.email = data.get('email', user.email)
user.save()

if old_email != user.email:
    send_verification_email(user)

# 初级思路：批量自动赋值
def basic_update(instance, fields, data):
    for field in fields:
        if field in data:
            # 这里的魔法：把字符串 field 变成实实在在的属性操作
            setattr(instance, field, data[field])

    # ❌ 这里的缺点：即使一个字都没改，也会触发一次全表的数据库写入操作。
    instance.save()

# 进阶：引入脏检查 (Dirty Check)
def efficient_update(instance, fields, data):
    has_updated = False
    for field in fields:
        if field in data:
            new_value = data[field]
            current_value = getattr(instance, field) # 先“看一眼”旧值

            # 只有真的变了，才动手
            if current_value != new_value:
                setattr(instance, field, new_value)
                has_updated = True

    if has_updated:
        # ✅ 精准写入：只更新 fields 中指定的列，不干扰其他字段。
        # 这也解决了 Race Condition（多个进程同时改不同字段导致互相覆盖）的问题。
        instance.save(update_fields=fields)

    return has_updated

# common/services.py

from typing import Tuple, Iterable
from django.db import models


def model_update(
    *,
    instance: models.Model,
    fields: Iterable[str],
    data: dict
) -> Tuple[models.Model, bool]:
    """
    通用模型更新服务

    只更新指定的字段，并且只在值实际改变时才保存。

    Args:
        instance: 要更新的模型实例
        fields: 允许更新的字段列表
        data: 包含新值的字典

    Returns:
        (更新后的实例, 是否有字段被修改)
    """
    has_updated = False

    # 遍历允许更新的字段
    for field in fields:
        # 检查 data 中是否包含该字段
        if field not in data:
            continue

        # 获取新值
        new_value = data[field]

        # 获取当前值
        current_value = getattr(instance, field)

        # 只在值确实改变时才更新
        if current_value != new_value:
            has_updated = True
            setattr(instance, field, new_value)

    # 只在有更新时才保存
    if has_updated:
        # 使用 update_fields 优化 SQL 查询，确保原子性
        instance.save(update_fields=fields)

    return instance, has_updated

# users/services.py

from styleguide_example.common.services import model_update
from styleguide_example.users.models import User


def user_update(*, user: User, data) -> User:
    """
    用户更新服务

    Args:
        user: 要更新的用户实例
        data: 包含更新字段的字典

    Returns:
        更新后的用户实例
    """
    # 定义没有副作用的字段
    non_side_effect_fields = ['first_name', 'last_name']

    # 使用通用 model_update 服务更新这些字段
    user, has_updated = model_update(
        instance=user,
        fields=non_side_effect_fields,
        data=data
    )

    # 在这里更新有副作用的字段
    # （例如，username 是根据 first_name 和 last_name 生成的）
    if has_updated:
        user.username = generate_username(
            first_name=user.first_name,
            last_name=user.last_name
        )
        user.save(update_fields=['username'])

    # ... 执行一些与用户相关的额外任务 ...

    # 例如：发送通知
    if has_updated:
        from users.tasks import user_profile_updated_task
        transaction.on_commit(
            lambda: user_profile_updated_task.delay(user.id)
        )

    return user

non_side_effect_fields = ['first_name', 'last_name']

user, has_updated = model_update(
    instance=user,
    fields=non_side_effect_fields,
    data=data
)

if has_updated:
    user.username = generate_username(...)
    user.save(update_fields=['username'])

def profile_update(*, profile: Profile, data: dict) -> Profile:
    """更新用户个人资料"""
    simple_fields = ['bio', 'location', 'website']

    profile, has_updated = model_update(
        instance=profile,
        fields=simple_fields,
        data=data
    )

    return profile

def product_update(*, product: Product, data: dict) -> Product:
    """更新产品信息"""
    # 验证
    if 'price' in data and data['price'] < 0:
        raise ValidationError("Price cannot be negative")

    # 更新无副作用字段
    basic_fields = ['name', 'description', 'price', 'stock']
    product, has_updated = model_update(
        instance=product,
        fields=basic_fields,
        data=data
    )

    # 处理副作用
    if has_updated and 'price' in data:
        # 价格变化时，记录历史
        PriceHistory.objects.create(
            product=product,
            old_price=product.price,
            new_price=data['price']
        )

    return product

def order_update(*, order: Order, data: dict, user: User) -> Order:
    """更新订单（带权限检查）"""
    # 根据用户角色决定可更新的字段
    if user.is_staff:
        allowed_fields = ['status', 'notes', 'shipping_address']
    else:
        allowed_fields = ['notes']

    order, has_updated = model_update(
        instance=order,
        fields=allowed_fields,
        data=data
    )

    # 状态变化时发送通知
    if has_updated and 'status' in data:
        from orders.tasks import order_status_changed_task
        transaction.on_commit(
            lambda: order_status_changed_task.delay(order.id)
        )

    return order

# common/tests/services/test_model_update.py

from django.test import TestCase
from users.tests.factories import UserFactory
from common.services import model_update


class ModelUpdateTests(TestCase):
    def test_updates_only_changed_fields(self):
        """测试只更新改变的字段"""
        user = UserFactory(first_name='John', last_name='Doe')

        data = {
            'first_name': 'Jane',  # 会改变
            'last_name': 'Doe',     # 不会改变
        }

        user, has_updated = model_update(
            instance=user,
            fields=['first_name', 'last_name'],
            data=data
        )

        self.assertTrue(has_updated)
        self.assertEqual(user.first_name, 'Jane')
        self.assertEqual(user.last_name, 'Doe')

    def test_returns_false_when_no_changes(self):
        """测试没有变化时返回 False"""
        user = UserFactory(first_name='John')

        data = {'first_name': 'John'}  # 相同的值

        user, has_updated = model_update(
            instance=user,
            fields=['first_name'],
            data=data
        )

        self.assertFalse(has_updated)

    def test_ignores_fields_not_in_data(self):
        """测试忽略不在 data 中的字段"""
        user = UserFactory(first_name='John', last_name='Doe')

        data = {'first_name': 'Jane'}  # 只包含 first_name

        user, has_updated = model_update(
            instance=user,
            fields=['first_name', 'last_name'],  # 但允许两个字段
            data=data
        )

        self.assertTrue(has_updated)
        self.assertEqual(user.first_name, 'Jane')
        self.assertEqual(user.last_name, 'Doe')  # 保持不变

    def test_ignores_fields_not_in_allowed_list(self):
        """测试忽略不在允许列表中的字段"""
        user = UserFactory(first_name='John', is_staff=False)

        data = {
            'first_name': 'Jane',
            'is_staff': True,  # 不在允许列表中
        }

        user, has_updated = model_update(
            instance=user,
            fields=['first_name'],  # 只允许 first_name
            data=data
        )

        self.assertTrue(has_updated)
        self.assertEqual(user.first_name, 'Jane')
        self.assertFalse(user.is_staff)  # 未被更新

[mypy]
python_version = 3.11
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
plugins =
    mypy_django_plugin.main,
    mypy_drf_plugin.main

[mypy.plugins.django-stubs]
django_settings_module = config.django.base

[mypy-*.migrations.*]
ignore_errors = True

[mypy-*.tests.*]
ignore_errors = True

[mypy-factory.*]
ignore_errors = True

from typing import Optional
from django.db import transaction
from users.models import User


@transaction.atomic
def user_create(
    *,
    email: str,
    first_name: str,
    last_name: str,
    password: Optional[str] = None
) -> User:
    """
    创建用户服务

    Args:
        email: 用户邮箱
        first_name: 名字
        last_name: 姓氏
        password: 密码（可选）

    Returns:
        创建的用户实例
    """
    user = User.objects.create_user(
        email=email,
        first_name=first_name,
        last_name=last_name,
    )

    if password is not None:
        user.set_password(password)
        user.save(update_fields=['password'])

    return user

from typing import Optional, Iterable
from django.db.models import QuerySet
from users.models import User


def user_list(
    *,
    filters: Optional[dict] = None
) -> QuerySet[User]:
    """
    用户列表选择器

    Args:
        filters: 过滤条件字典

    Returns:
        用户查询集
    """
    filters = filters or {}

    qs = User.objects.all()

    return qs.filter(**filters)


def user_get(*, user_id: int) -> User:
    """
    获取单个用户

    Args:
        user_id: 用户 ID

    Returns:
        用户实例

    Raises:
        User.DoesNotExist: 用户不存在时
    """
    return User.objects.get(id=user_id)

from typing import Any
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework import serializers, status


class UserCreateApi(APIView):
    class InputSerializer(serializers.Serializer):
        email = serializers.EmailField()
        first_name = serializers.CharField()
        last_name = serializers.CharField()
        password = serializers.CharField()

    def post(self, request: Request) -> Response:
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        from users.services import user_create

        user = user_create(**serializer.validated_data)

        return Response(status=status.HTTP_201_CREATED)

# 从关键部分开始
# services.py - 严格类型检查
def user_create(...) -> User:
    ...

# 逐步扩展到其他部分
# models.py, apis.py 等

[mypy-*.migrations.*]
ignore_errors = True

pip install django-stubs
pip install djangorestframework-stubs
pip install celery-stubs

# .github/workflows/ci.yml
- name: Run mypy
  run: mypy .