django-model-subscription

目录

• 动机
• 安装
• 用法
  • 装饰器
  • 使用AppConfig.ready设置订阅者
  • 使用自动发现设置订阅者
• 致谢
• 资源

动机

• 使用观察者模式通知订阅者关于django模型更改。
• 从模型的save方法解耦业务逻辑
• 支持批量操作（使用django signals不可用）
• 当settings.SUBSCRIPTION_RUN_EXTERNAL为False时使用no-op订阅者，这可以防止在测试、本地开发环境中模拟调用外部服务的订阅者。
• 显示实例更新后的更改，即比较初始状态和当前状态。

安装

$ pip install django-model-subscription

将 model_subscription 添加到您的 INSTALLED_APPS 中

INSTALLED_APPS = [
    ...,
    'model_subscription',
    ...
]

用法

使用 SubscriptionModelMixin 和 SubscriptionQuerySet

from model_subscription.mixin import SubscriptionModelMixin
from model_subscription.model import SubscriptionQuerySet


class TestModel(SubscriptionModelMixin, models.Model):
    name = models.CharField(max_length=255)

    objects = SubscriptionQuerySet.as_manager()

从 SubscriptionModel 基类派生。

from model_subscription.model import SubscriptionModel


class TestModel(SubscriptionModel):
    name = models.CharField(max_length=255)

创建订阅者。

• 使用 OperationType

import logging
from model_subscription.decorators import subscribe
from model_subscription.constants import OperationType

log = logging.getLogger(__name__)

@subscribe(OperationType.CREATE, TestModel)
def handle_create(instance):
    log.debug('Created {}'.format(instance.name))

• 直接使用 create_subscription（简洁版本）。

import logging
from model_subscription.decorators import create_subscription

log = logging.getLogger(__name__)

@create_subscription(TestModel)
def handle_create(instance):
    log.debug('Created {}'.format(instance.name))

装饰器

• subscribe：显式（需要有效的 OperationType）。

（创建、更新、删除）操作。

• create_subscription：订阅创建操作，即新实例。

@create_subscription(TestModel)
def handle_create(instance):
    log.debug('1. Created {}'.format(instance.name))

• update_subscription：订阅更新，还包括（changed_data）。

@update_subscription(TestModel)
def handle_update(instance, changed_data):
    log.debug('Updated {} {}'.format(instance.name, changed_data))

• delete_subscription：订阅删除操作

注意：实例的 pk 已经设置为 None。

@delete_subscription(TestModel)
def handle_delete(instance):
    log.debug('Deleted {}'.format(instance.name))

（批量创建、批量更新、批量删除）操作。

• bulk_create_subscription：订阅批量创建操作。

@bulk_create_subscription(TestModel)
def handle_bulk_create(instances):
    for instance in instances:
        log.debug('Bulk Created {}'.format(instance.name))

• bulk_update_subscription：订阅批量更新操作。

@bulk_update_subscription(TestModel)
def handle_bulk_update(instances):
    for instance in instances:
        log.debug('Updated {}'.format(instance.name))

• bulk_delete_subscription：订阅批量删除操作。

@bulk_delete_subscription(TestModel)
def handle_bulk_delete(instances):
    for instance in instances:
        log.debug('Deleted {}'.format(instance.name))

使用 AppConfig.ready 设置订阅者（推荐）。

更新您的 apps.py

from django.apps import AppConfig


class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        from myapp import subscriptions

使用自动发现设置订阅者。

默认情况下，settings.SUBSCRIPTION_AUTO_DISCOVER 设置为 False。

为了使用自动发现，这不建议，因为它会在模型使用的任何地方通知订阅者，例如 IPython 笔记本、外部脚本。

在您的 settings.py 中添加

SUBSCRIPTION_AUTO_DISCOVER = True

设置 SUBSCRIPTION_MODULE

注意：仅在 SUBSCRIPTION_AUTO_DISCOVER = True 时需要。

SUBSCRIPTION_MODULE  = 'subscription'

致谢

• django-lifecycle

如果您慷慨并想表示额外的感激之情

资源

• https://python-3-patterns-idioms-test.readthedocs.io/en/latest/Observer.html
• https://refactoring.guru/design-patterns/observer
• https://hackernoon.com/observer-vs-pub-sub-pattern-50d3b27f838c

待办事项

• 支持字段级别订阅。
• 支持基于类的订阅者，实现 __call__
• 扩展以包括自定义 OperationType。
• 添加对使用单个类管理多个操作的支持，例如 MyClass.update、MyClass.create。