Celery - importing models in tasks.py

Question

I'm having an issue getting access to models in my tasks.py

My goal is to send an email at various parts of the application (user registration, reset password etc..). To do this I pass the user id(s) to a celery task called 'send_email'.

@shared_task()
def send_email(sender_id=None, receiver_id=None, type=None, message=None):

    sender = User.objects.get(id=sender_id)
    receiver = User.objects.get(id=receiver_id)

    logger.info("Starting send email")
    Email.send_email(sender, receiver, type, message)
    logger.info("Finished send email")

The task then needs to use the id to retrieve the user and send them an email. This breaks down when trying to import the User model into the tasks.py file.

I receive an error

raise AppRegistryNotReady("Apps aren't loaded yet.") django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet.

Things I've tried

1. calling django.setup() at the top of tasks.py file - causes

   raise RuntimeError("populate() isn't reentrant") 
   

   causes the same error when put in send_email method as well. These were suggestions on other similar questions in SO

2. Importing model in 'send_email' method, allows the worker to start but causes the following error

   raise AppRegistryNotReady("Apps aren't loaded yet.") 
   

   This was another suggestion on a similar question in SO

3. Removing .delay when calling the 'send_email' function which works (with imports at top of tasks.py file or in send_email method) but as the task is no longer async it is not of any benefit but perhaps narrows down the issue?

Things of Note?

1. I use a custom user model that extends AbstractBaseUser, I have seen a number of github issues in celery relating to this but these were meant to be fixed in celery v3.1 or so I believe

2. I'm on celery v4.1, django 1.11.10, python 2.7 and am using RabbitMQ as Broker and running worker / server on a virtual env. I'm starting my worker using

   celery -A api worker -l info
   

   on a terminal window and then using pycharm's terminal to start server using

   python manage.py runserver
   

   so effectively there are 2 envs? Could this be the issue?

3. This may or not be related but in order to get my custom user model to work in my app/models.py I just have a single line that imports the User model otherwise I get a

   django.core.exceptions.ImproperlyConfigured: AUTH_USER_MODEL refers to model 'myApi.User' that has not been installed
   

4. I tried to set the auth model to 'myApi.user.User' (user being the folder where model is declared but get a

   Invalid model reference 'myApi.user.User'. String model references must be of the form 'app_label.ModelName'.
   

   so I'm guessing this is why the import is needed in myApi/models.py so that it can be picked up here?

Project Structure

├── api
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── celerySettings.py # my celery.py
├── db.sqlite3
├── myApi
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── tasks.py
│   ├── urls.py
│   ├── user
│   │   ├── __init__.py
│   │   ├── managers.py
│   │   ├── models.py
│   │   ├── serializers.py
│   │   ├── urls.py
│   │   └── views.py
│   ├── utils
│   │   └── Email.py
│   ├── views.py
├── manage.py
└── static

tasks.py

from __future__ import absolute_import, unicode_literals
from celery.schedules import crontab
from celery.task import periodic_task
from celery.utils.log import get_task_logger
from celery import shared_task

from celery import current_app

from .user.models import User
from .utils import Email

logger = get_task_logger(__name__)

@shared_task()
def send_email(sender_id=None, receiver_id=None, type=None, message=None):

    sender = User.objects.get(id=sender_id)
    receiver = User.objects.get(id=receiver_id)

    logger.info("Starting send email")
    Email.send_email(sender, receiver, type, message)
    logger.info("Finished send email")

settings.py

....
INSTALLED_APPS = [
    'rest_framework',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'corsheaders',
    'myApi',
    'celery',
    'rest_framework.authtoken',
    'rest_framework.renderers',
]

AUTH_USER_MODEL = 'myApi.User'
CELERY_IMPORTS = ('api.myApi.tasks')
....

celerySettings.py

from __future__ import absolute_import, unicode_literals
from django.conf import settings
import os
from celery import Celery


# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'api.api.settings')

app = Celery('api', broker='amqp://')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object(settings, namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

myApi/models.py

from user.models import User

myApi/admin.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

from django.contrib import admin

from user.models import User

admin.site.register(User)

api/wsgi.py

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "api.settings")

application = get_wsgi_application()

Any suggestions would be greatly appreciated. Also sorry for the long post, it is my first one so wasn't sure how much detail was needed.

Answer 1

I found my issue. In case it helps anyone else stuck on this I needed to add the line

sys.path.append(os.path.abspath('api'))

in my celerySettings.py for models to be picked up.

so it now looks like this

from __future__ import absolute_import, unicode_literals
from django.conf import settings
import os, sys
from celery import Celery

sys.path.append(os.path.abspath('api'))

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'api.api.settings')

app = Celery('api', broker='amqp://')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object(settings, namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

I then ran into another issue when actually trying to query the Database for my models locally. Celery was saying that my database tables did not exists, this was because it was creating a new database one folder above where the actual local database file was, to fix it I just had to change database name

"db.sqlite3"

to

os.path.join(os.path.dirname(__file__), "db.sqlite3")

in settings.py

which effectively changes it to

api/db.sqlite3

for Celery

Hopefully this helps someone else as I spent far too much time struggling this issue.