richleland / celery-test-project

A simple Django project using Celery for use in my quick start blog post.

Changed (Δ6.9 KB):

raw changeset »

__init__.py (null-size change)

manage.py (11 lines added, 0 lines removed)

people/__init__.py (null-size change)

people/admin.py (7 lines added, 0 lines removed)

people/models.py (23 lines added, 0 lines removed)

people/tasks.py (44 lines added, 0 lines removed)

people/tests.py (23 lines added, 0 lines removed)

people/views.py (6 lines added, 0 lines removed)

settings.py (89 lines added, 0 lines removed)

urls.py (9 lines added, 0 lines removed)

Up to file-list manage.py:

1
#!/usr/bin/env python
2
from django.core.management import execute_manager
3
try:
4
    import settings # Assumed to be in the same directory.
5
except ImportError:
6
    import sys
7
    sys.stderr.write("Error: Can't find the file 'settings.py' in the directory containing %r. It appears you've customized things.\nYou'll have to run django-admin.py, passing it your settings module.\n(If the file settings.py does indeed exist, it's causing an ImportError somehow.)\n" % __file__)
8
    sys.exit(1)
9
10
if __name__ == "__main__":
11
    execute_manager(settings)

Up to file-list people/admin.py:

1
from django.contrib import admin
2
from people.models import Person
3
4
class PersonAdmin(admin.ModelAdmin):
5
    list_display = ('__unicode__', 'date_of_birth', 'can_drink', 'call_delay')
6
7
admin.site.register(Person, PersonAdmin)

Up to file-list people/models.py:

1
from django.db import models
2
from django.core.urlresolvers import reverse
3
4
class Person(models.Model):
5
    date_of_birth = models.DateField()
6
    prefix = models.CharField(max_length=10, blank=True)
7
    first_name = models.CharField(max_length=255)
8
    middle_name = models.CharField(max_length=255, blank=True)
9
    last_name = models.CharField(max_length=255)
10
    suffix = models.CharField(max_length=10, blank=True)
11
    can_drink = models.BooleanField(default=False, editable=False)
12
    full_name = models.TextField(blank=True, editable=False)
13
    
14
    class Meta:
15
        verbose_name_plural = "people"
16
    
17
    def __unicode__(self):
18
        return self.full_name or self.last_name
19
    
20
    def call_delay(self):
21
        url = reverse('call-celery-delay', kwargs={'person_id': self.id})
22
        return '<a href="%s">Call Celery Delay</a>' % url
23
    call_delay.allow_tags = True

Up to file-list people/tasks.py:

1
from datetime import date, timedelta
2
from celery.task import Task, PeriodicTask
3
from celery.registry import tasks
4
from people.models import Person
5
6
class CanDrinkTask(Task):
7
    """
8
    A task that determines if a person is 21 years of age or older.
9
    """
10
    def run(self, person_id, **kwargs):
11
        logger = self.get_logger(**kwargs)
12
        logger.info("Running determine_can_drink task for person %s" % person_id)
13
    
14
        person = Person.objects.get(pk=person_id)
15
        now = date.today()
16
        diff = now - person.date_of_birth
17
        # i know, i know, this doesn't account for leap year
18
        age = diff.days / 365
19
        if age >= 21:
20
            person.can_drink = True
21
            person.save()
22
        else:
23
            person.can_drink = False
24
            person.save()
25
        return True
26
    
27
class FullNameTask(PeriodicTask):
28
    """
29
    A periodic task that concatenates fields to form a person's full name.
30
    """
31
    run_every = timedelta(seconds=60)
32
33
    def run(self, **kwargs):
34
        logger = self.get_logger(**kwargs)
35
        logger.info("Running full name task.")
36
        
37
        for person in Person.objects.all():
38
            person.full_name = " ".join([person.prefix, person.first_name,
39
                                         person.middle_name, person.last_name,
40
                                         person.suffix]).strip()
41
            person.save()
42
        return True
43
44
tasks.register(FullNameTask)

Up to file-list people/tests.py:

1
"""
2
This file demonstrates two different styles of tests (one doctest and one
3
unittest). These will both pass when you run "manage.py test".
4
5
Replace these with more appropriate tests for your application.
6
"""
7
8
from django.test import TestCase
9
10
class SimpleTest(TestCase):
11
    def test_basic_addition(self):
12
        """
13
        Tests that 1 + 1 always equals 2.
14
        """
15
        self.failUnlessEqual(1 + 1, 2)
16
17
__test__ = {"doctest": """
18
Another way to test that 1 + 1 is equal to 2.
19
20
>>> 1 + 1 == 2
21
True
22
"""}
23

Up to file-list people/views.py:

1
from django.http import HttpResponse
2
from people.tasks import CanDrinkTask
3
4
def call_celery_delay(request, person_id):
5
    CanDrinkTask.delay(person_id)
6
    return HttpResponse("Task set to execute.")

Up to file-list settings.py:

1
# Django settings for celeryproject project.
2
3
DEBUG = True
4
TEMPLATE_DEBUG = DEBUG
5
6
ADMINS = (
7
    # ('Your Name', 'your_email@domain.com'),
8
)
9
10
MANAGERS = ADMINS
11
12
DATABASE_ENGINE = 'sqlite3'           # 'postgresql_psycopg2', 'postgresql', 'mysql', 'sqlite3' or 'oracle'.
13
DATABASE_NAME = 'dev.db'             # Or path to database file if using sqlite3.
14
DATABASE_USER = ''             # Not used with sqlite3.
15
DATABASE_PASSWORD = ''         # Not used with sqlite3.
16
DATABASE_HOST = ''             # Set to empty string for localhost. Not used with sqlite3.
17
DATABASE_PORT = ''             # Set to empty string for default. Not used with sqlite3.
18
19
# Local time zone for this installation. Choices can be found here:
20
# http://en.wikipedia.org/wiki/List_of_tz_zones_by_name
21
# although not all choices may be available on all operating systems.
22
# If running in a Windows environment this must be set to the same as your
23
# system time zone.
24
TIME_ZONE = 'America/New_York'
25
26
# Language code for this installation. All choices can be found here:
27
# http://www.i18nguy.com/unicode/language-identifiers.html
28
LANGUAGE_CODE = 'en-us'
29
30
SITE_ID = 1
31
32
# If you set this to False, Django will make some optimizations so as not
33
# to load the internationalization machinery.
34
USE_I18N = True
35
36
# Absolute path to the directory that holds media.
37
# Example: "/home/media/media.lawrence.com/"
38
MEDIA_ROOT = ''
39
40
# URL that handles the media served from MEDIA_ROOT. Make sure to use a
41
# trailing slash if there is a path component (optional in other cases).
42
# Examples: "http://media.lawrence.com", "http://example.com/media/"
43
MEDIA_URL = ''
44
45
# URL prefix for admin media -- CSS, JavaScript and images. Make sure to use a
46
# trailing slash.
47
# Examples: "http://people.com/media/", "/media/".
48
ADMIN_MEDIA_PREFIX = '/media/'
49
50
# Make this unique, and don't share it with anybody.
51
SECRET_KEY = '^2yh8+kwlgv9q=11=*oy$tnod7-66fxjm-l6&3sdn8*wgy2^%('
52
53
# List of callables that know how to import templates from various sources.
54
TEMPLATE_LOADERS = (
55
    'django.template.loaders.filesystem.load_template_source',
56
    'django.template.loaders.app_directories.load_template_source',
57
#     'django.template.loaders.eggs.load_template_source',
58
)
59
60
MIDDLEWARE_CLASSES = (
61
    'django.middleware.common.CommonMiddleware',
62
    'django.contrib.sessions.middleware.SessionMiddleware',
63
    'django.contrib.auth.middleware.AuthenticationMiddleware',
64
)
65
66
ROOT_URLCONF = 'urls'
67
68
TEMPLATE_DIRS = (
69
    # Put strings here, like "/home/html/django_templates" or "C:/www/django/templates".
70
    # Always use forward slashes, even on Windows.
71
    # Don't forget to use absolute paths, not relative paths.
72
)
73
74
INSTALLED_APPS = (
75
    'django.contrib.auth',
76
    'django.contrib.contenttypes',
77
    'django.contrib.sessions',
78
    'django.contrib.sites',
79
    'django.contrib.admin',
80
    'people',
81
    'celery',
82
)
83
84
CELERY_BACKEND = "database"
85
BROKER_HOST = "localhost"
86
BROKER_PORT = 5672
87
BROKER_USER = "myusername"
88
BROKER_PASSWORD = "mypassword"
89
BROKER_VHOST = "myvhost"

Up to file-list urls.py:

1
from django.conf.urls.defaults import *
2
from django.contrib import admin
3
4
admin.autodiscover()
5
6
urlpatterns = patterns('',
7
    (r'^admin/', include(admin.site.urls)),
8
    url(r'^people/delay/(?P<person_id>\d+)/$', 'people.views.call_celery_delay', name='call-celery-delay'),
9
)