如何允许在 Django 中重新定义函数？

Question

我有一个“核心”Django 产品，其中包含常见任务的默认实现，但我希望允许重新定义该实现（或自定义，如果这样更容易的话）。

例如在核心产品中，我可能有一个允许用户单击按钮重新发送“所有通知”的视图：

# in core/views.py
... imports etc...
from core.tasks import resend_notifications

def handle_user_resend_request(request, user_id):
    user = get_object_or_404(id=user_id)

    if request.method == 'POST':
        for follower in user.followers:
            resend_notifications(follower.id)

    ... etc etc ...


# in core/tasks.py
... imports etc...

def resend_notifications(id):
    send_email(User.objects.get(id=id))

然后在此产品的某些部署中，“重新发送通知”可能需要如下所示：

# in customer_specific/tasks.py
... imports etc ...

def resend_notifications(id):
    person = User.objects.get(id=id)
    if '@super-hack.email.com' in person.email:
        # This is not a real email, send via the magic portal
        send_via_magic(person)
    else:
        send_email(person)
     # and send via fax for good measure
    send_fax(person)

如何让views.py 文件中的resend_notifications 函数指向customer_specific 版本？

我应该在 Django 配置中定义它并以这种方式共享访问权限吗？如果这些任务实际上是 Celery 任务怎么办？

注意：我的任务实际上被定义为 Celery 任务（我删除了这个额外的细节，因为我认为这个问题更笼统）。我尝试过使用自定义装饰器标签来改变全局对象，但出于多种原因，这绝对不是可行的方法。

PS：我觉得这是一个依赖注入问题，但这在Django中并不常见。

Answer 1

在类似的情况下，我最终选择了这样的解决方案——我将它放在应用程序中的 Organization 模型上（相当于 GitHub 组织）。

@property
def forms(self):
    if self.ldap:
        from portal.ldap import forms
    else:
        from portal.users import forms

    return forms

如果经过身份验证的用户所属的组织配置了 LDAP，我本质上想使用不同的表单类 - 因此创建/邀请用户表单需要不同。

然后我在适当的视图中覆盖get_form_class，如下所示：

def get_form_class(self):
    return self.request.user.organization.forms.CreateUserForm

我想您可能想在您的场景中做类似的事情，将您的函数包装在一个代理抽象中，该抽象决定要使用哪个版本 - 无论是基于环境变量、设置还是请求。

Answer 2

这最终通过可以由部署配置重新配置的 Django 设置对象解决。它的灵感主要来自这里的技术：settings.py from django-rest-framework。

例如，我的项目中有这样一个设置文件：

yourproject/settings.py

"""
Settings for <YOUR PROJECT> are all namespaced in the YOUR_PROJECT config option.
For example your project's config file (usually called `settings.py` or 'production.py') might look like this:

YOUR_PROJECT = {
    'PROCESS_TASK': (
        'your_project.tasks.process_task',
    )
}

This module provides the `yourproject_settings` object, that is used
to access settings, checking for user settings first, then falling
back to the defaults.
"""
# This file was effectively borrow from https://github.com/tomchristie/django-rest-framework/blob/8385ae42c06b8e68a714cb67b7f0766afe316883/rest_framework/settings.py

from __future__ import unicode_literals
from django.conf import settings
from django.utils.module_loading import import_string


DEFAULTS = {
    'RESEND_NOTIFICATIONS_TASK': 'core.tasks.resend_notifications',
}


# List of settings that may be in string import notation.
IMPORT_STRINGS = (
    'RESEND_NOTIFICATIONS_TASK',
)


MANDATORY_SETTINGS = (
    'RESEND_NOTIFICATIONS_TASK',
)


def perform_import(val, setting_name):
    """
    If the given setting is a string import notation,
    then perform the necessary import or imports.
    """
    if val is None:
        return None
    if callable(val):
        return val
    if isinstance(val, (list, tuple)):
        return [perform_import(item, setting_name) for item in val]

    try:
        return import_string(val)
    except (ImportError, AttributeError) as e:
        msg = "Could not import '%s' for setting '%s'. %s: %s." % (val, setting_name, e.__class__.__name__, e)
        raise ImportError(msg)


class YourProjectSettings(object):
    """
    A settings object, that allows settings to be accessed as properties.
    For example:

        from your_project.settings import yourproject_settings as the_settings
        print(the_settings.RESEND_NOTIFICATIONS_TASK)

    Any setting with string import paths will be automatically resolved
    and return the class, rather than the string literal.
    """
    namespace = 'YOUR_PROJECT'

    def __init__(self, mandatory=None, defaults=None, import_strings=None):
        self.mandatory = mandatory or MANDATORY_SETTINGS
        self.defaults = defaults or DEFAULTS
        self.import_strings = import_strings or IMPORT_STRINGS

        self.__check_settings()

    @property
    def user_settings(self):
        if not hasattr(self, '_user_settings'):
            self._user_settings = getattr(settings, self.__class__.namespace, {})
        return self._user_settings

    def __getattr__(self, attr):
        if attr not in self.defaults and attr not in self.mandatory:
            raise AttributeError("Invalid Pyrite setting: '%s'" % attr)

        try:
            # Check if present in user settings
            val = self.user_settings[attr]
        except KeyError:
            # Fall back to defaults
            val = self.defaults[attr]

        # Coerce import strings into classes
        if attr in self.import_strings:
            val = perform_import(val, attr)

        # Cache the result
        setattr(self, attr, val)
        return val

    def __check_settings(self):
        for setting in self.mandatory:
            if setting not in self.user_settings:
                raise RuntimeError(
                    'The "{}" setting is required as part of the configuration for "{}", but has not been supplied.'.format(
                    setting, self.__class__.namespace))


yourproject_settings = YourProjectSettings(MANDATORY_SETTINGS, DEFAULTS, IMPORT_STRINGS)

这让我可以：

• 使用默认值（即'core.tasks.resend_notications'）；或

• 在我的配置文件中重新定义绑定：

  site_config/special.py

  ... other django settings like DB / DEBUG / Static files etc
  
  YOUR_PROJECT = {
      'RESEND_NOTIFICATIONS_TASK': 'customer_specific.tasks.resend_notifications',
  }
  
  ... etc. ...
  

然后在我的视图函数中，我通过设置访问正确的函数：

core/views.py

... imports etc...
from yourproject.settings import yourproject_settings as my_settings

def handle_user_resend_request(request, user_id):
    user = get_object_or_404(id=user_id)

    if request.method == 'POST':
        for follower in user.followers:
            my_settings.RESEND_NOTIFICATIONS_TASK(follower.id)

    ... etc etc ...