Environment Setup

• OS: Ubuntu
• Editor: PyCharm
• Python: 3.6 (bundled with Celery 4.3.0)
• Dependencies: Django 3.0.8

Celery 4 can drive scheduled jobs in Django without extra plugins like djcelery. Using the native celery beat keeps the setup lightweight. Trade-offs include losing plugin-based task dashboards and simplified task management, which is acceptable when changes to schedules are infrequent.

Project Initialization

django-admin startproject Newsite
cd Newsite
python3 manage.py startapp aggregator
touch ./Newsite/celery_app.py ./aggregator/jobs.py start_worker.sh

Resulting layout:

Newsite/
├── aggregator/
│   ├── admin.py
│   ├── apps.py
│   ├── __init__.py
│   ├── migrations/
│   │   └── __init__.py
│   ├── models.py
│   ├── jobs.py
│   ├── tests.py
│   └── views.py
├── manage.py
├── start_worker.sh
└── Newsite/
    ├── asgi.py
    ├── celery_app.py
    ├── __init__.py
    ├── settings.py
    ├── urls.py
    └── wsgi.py

aggregator handles news collection; Newsite is the project container.

Data Model

For persistnet ranking history, define a model to store entries:

# aggregator/models.py
from django.db import models

class RankItem(models.Model):
    INACTIVE = 0
    ACTIVE = 1
    STATE_CHOICES = ((INACTIVE, 'Inactive'), (ACTIVE, 'Active'))

    source_name = models.CharField('Source', max_length=256, null=True)
    category = models.CharField('Category', max_length=256, null=True)
    payload = models.TextField('Data', null=True)
    created = models.DateTimeField(auto_now_add=True)
    modified = models.DateTimeField(auto_now=True)
    state = models.SmallIntegerField(choices=STATE_CHOICES, default=ACTIVE)
    rank_order = models.SmallIntegerField('Order', default=0)

    def __str__(self):
        return f"{self.source_name}{self.category}"

    class Meta:
        ordering = ['-rank_order']

View Layer

# aggregator/views.py
from django.shortcuts import render
from .models import RankItem

def show_rankings(request):
    items = RankItem.objects.filter(state=RankItem.ACTIVE)
    context = {'title': 'Trending News', 'items': items}
    return render(request, 'aggregator/rankings.html', context)

Task Definition

# aggregator/jobs.py
import sys
import os
import django
from concurrent.futures import ThreadPoolExecutor

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Newsite.settings')
django.setup()

from celery import shared_task
from aggregator.scrapers import fetch_github_trending
from aggregator.models import RankItem

@shared_task
def refresh_rankings():
    scrapers = [fetch_github_trending]

    def handle_result(fut):
        outcome = fut.result()
        src = outcome.get('source_name', '')
        cat = outcome.get('category', '')
        data = outcome.get('payload', '')
        if src:
            rec = RankItem.objects.filter(source_name=src).first()
            if not rec:
                RankItem.objects.create(source_name=src, category=cat, payload=data)
            else:
                rec.payload = data
                rec.save()

    with ThreadPoolExecutor(max_workers=4) as executor:
        for job in scrapers:
            executor.submit(job).add_done_callback(handle_result)
    print('Refresh completed')

Example Scraper

# aggregator/scrapers.py
import requests
from lxml import etree
import sys

def fetch_github_trending():
    endpoint = 'https://github.com/trending'
    hdr = {'Host': 'github.com', 'Referer': 'https://github.com/explore'}
    resp = requests.get(endpoint, headers=hdr, timeout=5)
    entries = []
    if resp:
        doc = etree.HTML(resp.text)
        articles = doc.xpath("//article[@class='Box-row']")
        for art in articles:
            title = art.xpath('string(./h1/a)').strip()
            link = 'https://github.com' + art.xpath('./h1/a/@href')[0]
            summary = art.xpath('string(./p)').strip()
            entries.append({'title': f'{title}---{summary}', 'url': link})
    return {
        'source_name': 'GitHub',
        'category': 'Trending',
        'scraper': sys._getframe().f_code.co_name,
        'payload': entries
    }

URL Configuration

# aggregator/urls.py
from django.urls import path
from . import views

app_name = 'aggregator'
urlpatterns = [
    path('rankings', views.show_rankings, name='show_rankings'),
]

# Newsite/urls.py
from django.urls import path, include
from django.contrib import admin

urlpatterns = [
    path('admin/', admin.site.urls),
    path('news/', include('aggregator.urls', namespace='aggregator')),
]

Celery Integration

Celery App Factory

# Newsite/celery_app.py
import os
from celery import Celery
from datetime import timedelta

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Newsite.settings')
celery_instance = Celery('NewsiteWorker')
celery_instance.config_from_object('django.conf:settings', namespace='CELERY')
celery_instance.autodiscover_tasks()

celery_instance.conf.update(
    beat_schedule={
        'update-rankings': {
            'task': 'aggregator.jobs.refresh_rankings',
            'schedule': timedelta(seconds=600),
        }
    }
)

Django Import Hook

# Newsite/__init__.py
from .celery_app import celery_instance as celery_app
__all__ = ('celery_app',)

Setttings Adjustments

# Newsite/settings.py
import os

REDIS_HOST = os.environ.get('REDIS_ADDR', 'localhost')
REDIS_PORT = 6379
REDIS_DB = 0

CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'UTC'
CELERY_BROKER_URL = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}'
CELERY_RESULT_BACKEND = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

Scheduler Activation

Launch worker and beat together:

# start_worker.sh
export REDIS_ADDR=127.0.0.1
celery -A Newsite worker -l info -B -f /var/log/celery.log

Avoid running as root. Monitor /var/log/celery.log to verify periodic execution of refresh_rankings.