project_name/
├── app_name/
│   ├── __init__.py
│   ├── models.py
│   ├── services.py
│   ├── selectors.py
│   ├── apis.py
│   └── tests/
│       ├── __init__.py
│       ├── factories.py           # Factory Boy 工厂定义
│       ├── models/
│       │   ├── __init__.py
│       │   └── test_some_model_name.py
│       ├── selectors/
│       │   ├── __init__.py
│       │   └── test_some_selector_name.py
│       └── services/
│           ├── __init__.py
│           └── test_some_service_name.py
└── __init__.py

test_the_name_of_the_thing_that_is_tested.py

# 被测试的服务
def a_very_neat_service(*args, **kwargs):
    pass

project_name/app_name/tests/services/test_a_very_neat_service.py

class TheNameOfTheThingThatIsTestedTests(TestCase):
    pass

# tests/services/test_a_very_neat_service.py

from django.test import TestCase


class AVeryNeatServiceTests(TestCase):
    """
    测试 a_very_neat_service 服务
    """

    def test_service_does_something_correctly(self):
        # 测试实现
        pass

    def test_service_handles_edge_case(self):
        # 测试实现
        pass

project_name/common/utils.py

project_name/common/tests/test_utils.py

project_name/common/utils/
├── __init__.py
├── files.py
├── strings.py
└── dates.py

project_name/common/tests/utils/
├── __init__.py
├── test_files.py
├── test_strings.py
└── test_dates.py

@pytest.fixture
def active_user():
    # 在 fixture 中利用 Factory 快速创建复杂对象
    return UserFactory(is_active=True, has_profile=True)

def test_login(client, active_user):
    # 结合使用：client 是 pytest fixture，active_user 是由 factory 生成的 fixture
    client.force_login(active_user)
    ...

# ❌ 不推荐：在每个测试中手动创建对象
class UserServiceTests(TestCase):
    def test_user_can_update_profile(self):
        user = User.objects.create(
            email='test@example.com',
            first_name='John',
            last_name='Doe',
            is_active=True,
            # ... 许多其他字段 ...
        )
        # 测试逻辑

    def test_user_can_delete_account(self):
        user = User.objects.create(
            email='test2@example.com',
            first_name='Jane',
            last_name='Smith',
            is_active=True,
            # ... 又要重复许多字段 ...
        )
        # 测试逻辑

# ✅ 推荐：使用 Factory
class UserServiceTests(TestCase):
    def test_user_can_update_profile(self):
        user = UserFactory()
        # 测试逻辑 - 关注点清晰

    def test_user_can_delete_account(self):
        user = UserFactory()
        # 测试逻辑

    def test_inactive_user_cannot_login(self):
        # 只覆盖需要的字段
        user = UserFactory(is_active=False)
        # 测试逻辑

pip install factory_boy

# app_name/tests/factories.py

import factory
from faker import Faker

from app_name.models import User, Profile


fake = Faker()


class UserFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = User

    email = factory.Sequence(lambda n: f'user{n}@example.com')
    first_name = factory.LazyFunction(lambda: fake.first_name())
    last_name = factory.LazyFunction(lambda: fake.last_name())
    is_active = True
    is_staff = False


class ProfileFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = Profile

    user = factory.SubFactory(UserFactory)
    bio = factory.LazyFunction(lambda: fake.text(max_nb_chars=200))
    avatar = factory.django.ImageField(color='blue')

# 创建一个用户（所有字段使用默认值）
user = UserFactory()

# 覆盖特定字段
admin_user = UserFactory(is_staff=True, email='admin@example.com')

# 创建多个对象
users = UserFactory.create_batch(5)

# 创建带关联的对象
profile = ProfileFactory()  # 自动创建关联的 user

# 只构建对象，不保存到数据库
user = UserFactory.build()

class UserFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = User

    email = factory.Sequence(lambda n: f'user{n}@example.com')
    # 默认是普通活跃用户
    is_active = True
    is_staff = False

    class Params:
        # 定义 "admin" 特征：一旦开启，自动设置相关的一组字段
        admin = factory.Trait(
            is_staff=True,
            is_superuser=True,
        )
        # 定义 "inactive" 特征
        inactive = factory.Trait(
            is_active=False,
        )

# 🔥 威力展示：
# 给我来个管理员！
admin_user = UserFactory(admin=True)

# 给我来个被封禁的用户！
banned_user = UserFactory(inactive=True)

class OrderFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = Order

    user = factory.SubFactory(UserFactory)
    status = 'pending'

    # 定义一个名为 "items" 的钩子
    @factory.post_generation
    def items(self, create, extracted, **kwargs):
        if not create:
            # 如果只是 build() 而不存库，没法创建关联对象，直接返回
            return

        if extracted:
            # 情况 A：调用者指定了 items -> OrderFactory(items=[item1, item2])
            for item in extracted:
                self.items.add(item)
        else:
            # 情况 B：调用者啥都没给 -> 自动送给你 3 个默认商品！
            OrderItemFactory.create_batch(3, order=self)

# 🔥 威力展示：
# 自动创建一个订单，且它已经带了 3 个商品，可以直接拿去测计算总价逻辑！
order = OrderFactory()
assert order.items.count() == 3

from faker import Faker

# 支持多语言，随机生成中文或英文数据
fake = Faker(['en_US', 'zh_CN'])


class ArticleFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = Article

    # 随机生成一句像样的话作为标题
    title = factory.LazyFunction(lambda: fake.sentence())
    # 随机生成一篇 2000 字的文章
    content = factory.LazyFunction(lambda: fake.text(max_nb_chars=2000))
    # 自动关联一个作者
    author = factory.SubFactory(UserFactory)
    # 随机生成一个今年的发布时间
    published_at = factory.LazyFunction(lambda: fake.date_time_this_year())

# tasks.py
@shared_task
def process_order(order_id):
    # 😱 糟糕：ORM 操作、业务规则、第三方调用全混在一起
    order = Order.objects.get(id=order_id)
    if order.status != 'paid':
        return

    #复杂的计算逻辑...
    order.amount = order.items.aggregate(Sum('price'))
    order.save()

    # 发送邮件逻辑...
    send_mail(f"Order {order.id} processed", ...)

# services.py
def order_process(order: Order):
    # 真正的业务逻辑在这里，纯粹的 Python 函数
    ...

# tasks.py
@shared_task
def process_order(order_id):
    # 😍 完美：只负责“把盘子端给厨师”
    order = Order.objects.get(id=order_id)
    order_process(order)

# emails/services.py

from django.db import transaction
from django.core.mail import EmailMultiAlternatives
from django.utils import timezone

from styleguide_example.core.exceptions import ApplicationError
from styleguide_example.common.services import model_update
from styleguide_example.emails.models import Email


@transaction.atomic
def email_send(email: Email) -> Email:
    """
    发送邮件服务

    这里包含业务逻辑：
    - 验证邮件状态
    - 实际发送邮件
    - 更新邮件状态
    """
    # 业务规则：只能发送状态为 SENDING 的邮件
    if email.status != Email.Status.SENDING:
        raise ApplicationError(
            f"Cannot send non-ready emails. Current status is {email.status}"
        )

    # 准备邮件内容
    subject = email.subject
    from_email = "styleguide-example@hacksoft.io"
    to = email.to
    html = email.html
    plain_text = email.plain_text

    # 发送邮件
    msg = EmailMultiAlternatives(subject, plain_text, from_email, [to])
    msg.attach_alternative(html, "text/html")
    msg.send()

    # 更新邮件状态
    email, _ = model_update(
        instance=email,
        fields=["status", "sent_at"],
        data={
            "status": Email.Status.SENT,
            "sent_at": timezone.now()
        }
    )

    return email

# emails/tasks.py

from celery import shared_task

from styleguide_example.emails.models import Email


@shared_task
def email_send(email_id):
    """
    邮件发送任务

    这个任务很薄，只做两件事：
    1. 获取数据
    2. 调用服务
    """
    # 1. 获取需要的数据
    email = Email.objects.get(id=email_id)

    # 2. 调用服务（在函数体内导入，避免循环导入）
    from styleguide_example.emails.services import email_send

    email_send(email)

# ✅ 推荐：在函数体内导入
@shared_task
def email_send(email_id):
    from styleguide_example.emails.services import email_send
    # ...

# ❌ 不推荐：在模块顶部导入
from styleguide_example.emails.services import email_send

@shared_task
def email_send_task(email_id):
    # ...

# users/services.py

from django.db import transaction

from styleguide_example.users.models import User
# 注意：我们在模块级导入任务，并给它加 _task 后缀
from styleguide_example.emails.tasks import email_send as email_send_task


@transaction.atomic
def user_complete_onboarding(user: User) -> User:
    """
    用户完成入职流程
    """
    # ... 一些业务逻辑代码 ...

    # 准备邮件
    email = email_get_onboarding_template(user=user)

    # 事务提交后，触发异步任务
    transaction.on_commit(lambda: email_send_task.delay(email.id))

    return user

from styleguide_example.emails.tasks import email_send as email_send_task

transaction.on_commit(lambda: email_send_task.delay(email.id))

# emails/tasks.py

from celery import shared_task
from celery.utils.log import get_task_logger

from styleguide_example.emails.models import Email


logger = get_task_logger(__name__)


def _email_send_failure(self, exc, task_id, args, kwargs, einfo):
    """
    任务失败回调

    当所有重试都失败后，这个函数会被调用
    """
    email_id = args[0]
    email = Email.objects.get(id=email_id)

    # 调用服务来处理失败情况
    from styleguide_example.emails.services import email_failed
    email_failed(email)


@shared_task(bind=True, on_failure=_email_send_failure)
def email_send(self, email_id):
    """
    带重试机制的邮件发送任务
    """
    email = Email.objects.get(id=email_id)

    from styleguide_example.emails.services import email_send

    try:
        email_send(email)
    except Exception as exc:
        # 记录警告日志
        logger.warning(f"Exception occurred while sending email: {exc}")

        # 重试任务
        # https://docs.celeryq.dev/en/stable/userguide/tasks.html#retrying
        self.retry(exc=exc, countdown=5)

@shared_task(bind=True, on_failure=_email_send_failure)
def email_send(self, email_id):
    # self 现在是任务实例

self.retry(exc=exc, countdown=5)

def _email_send_failure(self, exc, task_id, args, kwargs, einfo):
    # 处理最终失败

@shared_task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),  # 自动重试这些异常
    retry_kwargs={'max_retries': 5},  # 最多重试 5 次
    retry_backoff=True,  # 使用指数退避
    retry_backoff_max=600,  # 最大退避时间 10 分钟
    retry_jitter=True,  # 添加随机抖动
)
def send_notification(self, user_id, message):
    """
    发送通知，带高级重试机制
    """
    user = User.objects.get(id=user_id)

    from notifications.services import notification_send

    try:
        notification_send(user=user, message=message)
    except RateLimitError as exc:
        # 遇到限流错误，等待更长时间再重试
        raise self.retry(exc=exc, countdown=60 * 5)  # 5 分钟后重试

# project/tasks/celery.py

import os
from celery import Celery

# 设置 Django 设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.django.base')

app = Celery('project')

# 从 Django 设置加载配置，使用 CELERY_ 前缀
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现所有已安装应用的 tasks.py
app.autodiscover_tasks()

# config/settings/celery.py

from config.env import env

CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND = env('CELERY_RESULT_BACKEND', default='redis://localhost:6379/0')

CELERY_TIMEZONE = 'UTC'

# 任务跟踪
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60  # 30 分钟

# 结果过期时间
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 1 天

# 序列化
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# 其他配置
CELERY_TASK_ALWAYS_EAGER = False  # 测试时设为 True
CELERY_TASK_EAGER_PROPAGATES = True

app_name/
├── models.py
├── services.py
├── tasks.py          # 所有任务在一个文件中
└── tests/

app_name/
├── models.py
├── services.py
├── tasks/
│   ├── __init__.py   # 导入所有任务，让 Celery 自动发现
│   ├── domain_a.py   # 领域 A 的任务
│   └── domain_b.py   # 领域 B 的任务
└── tests/

# 从子模块导入所有任务，让 Celery 能够发现它们
from .domain_a import *  # noqa
from .domain_b import *  # noqa

emails/tasks/
├── __init__.py
├── sending.py       # 发送相关任务
├── templates.py     # 模板处理任务
└── analytics.py     # 分析相关任务

notifications/tasks/
├── __init__.py
├── sms.py          # SMS 通知任务
├── push.py         # Push 通知任务
└── email.py        # Email 通知任务

# tasks/management/commands/setup_periodic_tasks.py

from django.core.management.base import BaseCommand
from django.db import transaction

from django_celery_beat.models import IntervalSchedule, CrontabSchedule, PeriodicTask

from project.app.tasks import some_periodic_task


class Command(BaseCommand):
    help = f"""
    设置 celery beat 周期性任务。

    将创建以下任务：

    - {some_periodic_task.name}
    """

    @transaction.atomic
    def handle(self, *args, **kwargs):
        print('删除所有周期性任务和计划...\n')

        # 清理现有的
        IntervalSchedule.objects.all().delete()
        CrontabSchedule.objects.all().delete()
        PeriodicTask.objects.all().delete()

        # 定义所有周期性任务
        periodic_tasks_data = [
            {
                'task': some_periodic_task,
                'name': 'Do some periodic stuff',
                # https://crontab.guru/#15_*_*_*_*
                'cron': {
                    'minute': '15',
                    'hour': '*',
                    'day_of_week': '*',
                    'day_of_month': '*',
                    'month_of_year': '*',
                },
                'enabled': True
            },
        ]

        # 创建任务
        for periodic_task in periodic_tasks_data:
            print(f'设置 {periodic_task["task"].name}')

            cron = CrontabSchedule.objects.create(
                **periodic_task['cron']
            )

            PeriodicTask.objects.create(
                name=periodic_task['name'],
                task=periodic_task['task'].name,
                crontab=cron,
                enabled=periodic_task['enabled']
            )

        print('\n✅ 周期性任务设置完成！')

# 在部署脚本中
python manage.py migrate
python manage.py setup_periodic_tasks  # 设置周期性任务
python manage.py collectstatic --noinput

# https://crontab.guru/#15_*_*_*_*
'cron': {
    'minute': '15',
    'hour': '*',
    'day_of_week': '*',
    'day_of_month': '*',
    'month_of_year': '*',
},

periodic_tasks_data = [
    {
        'task': send_daily_digest,
        'name': 'Send daily digest emails',
        # 每天早上 9 点
        # https://crontab.guru/#0_9_*_*_*
        'cron': {
            'minute': '0',
            'hour': '9',
            'day_of_week': '*',
            'day_of_month': '*',
            'month_of_year': '*',
        },
        'enabled': True
    },
    {
        'task': cleanup_old_sessions,
        'name': 'Cleanup old sessions',
        # 每周日凌晨 2 点
        # https://crontab.guru/#0_2_*_*_0
        'cron': {
            'minute': '0',
            'hour': '2',
            'day_of_week': '0',
            'day_of_month': '*',
            'month_of_year': '*',
        },
        'enabled': True
    },
    {
        'task': generate_monthly_report,
        'name': 'Generate monthly reports',
        # 每月1号早上 6 点
        # https://crontab.guru/#0_6_1_*_*
        'cron': {
            'minute': '0',
            'hour': '6',
            'day_of_week': '*',
            'day_of_month': '1',
            'month_of_year': '*',
        },
        'enabled': True
    },
]

from celery import chain

# 就像多米诺骨牌，一个推倒一个
workflow = chain(
    process_upload.s(file_id),  # 返回处理后的路径
    validate_file.s(),          # 接收路径并验证
    import_data.s(),            # 接收验证结果并入库
    send_notification.s()       # 发送成功通知
)
workflow.apply_async()

from celery import group

# 3 个 worker 如果空闲，会同时开始工作
job = group([
    resize_image.s(image_id, size='small'),
    resize_image.s(image_id, size='medium'),
    resize_image.s(image_id, size='large'),
])
result = job.apply_async()

from celery import chord

# 并行执行 header，完成后执行 callback
callback = generate_report.s()
header = group([
    fetch_data_from_api_a.s(),
    fetch_data_from_api_b.s(),
    fetch_data_from_api_c.s(),
])
chord(header)(callback)

from celery import group

# 对列表中的每个 ID 执行任务
user_ids = [1, 2, 3, 4, 5]
# 将任务分发给所有 user_ids
job = group(send_welcome_email.s(uid) for uid in user_ids)
result = job.apply_async()