Skip to content

tools

django_spire.celery.tools

discover_celery_tasks

Custom task discovery to look for specific patterns or avoid errors.

Source code in django_spire/celery/tools.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
@celeryd_after_setup.connect
def discover_celery_tasks(sender, instance, **kwargs):
    """
    Custom task discovery to look for specific patterns or avoid errors.
    """
    from django.apps import apps
    import importlib

    for app_name in [config.name for config in apps.get_app_configs()]:
        try:
            module_name = f'{app_name}.celery.tasks'
            importlib.import_module(module_name)
        except ImportError:
            pass