Django Daily Record Check with Notifications

I'll guide you through implementing a daily check for record creation with notifications in your Django app. Here's a step-by-step approach:

1. Setting up a Scheduled Task

For Django, you can use Celery with Celery Beat for scheduled tasks instead of traditional cron jobs:

Install required packages:

BASHCopied
pip install celery django-celery-beat redis

Configure Celery in your Django project:

Create a celery.py file in your project directory:

PYTHONCopied
# myproject/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

Update your __init__.py file:

PYTHONCopied
# myproject/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Add Celery settings to settings.py:

PYTHONCopied
# settings.py
INSTALLED_APPS = [
    # ... other apps
    'django_celery_beat',
]

# Celery settings
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'  # Set to your timezone

2. Create the Task to Check Records

Create a task in your app's tasks.py file:

PYTHONCopied
# yourapp/tasks.py
from celery import shared_task
from django.utils import timezone
from .models import YourRecord  # Replace with your actual model
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

@shared_task
def check_daily_record():
    # Check if a record was created today
    today = timezone.now().date()
    record_exists = YourRecord.objects.filter(
        created_at__date=today
    ).exists()
    
    if not record_exists:
        # Send notification via WebSockets
        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)(
            "notifications",  # Group name
            {
                "type": "send_notification",
                "message": "Time to create your daily record!"
            }
        )
    
    return record_exists

3. Schedule the Task

You can schedule this task using Django admin after setting up Celery Beat:

1. Run migrations: python manage.py migrate
2. Create a periodic task in Django admin at /admin/django_celery_beat/periodictask/
3. Set it to run daily at 11:00 AM

4. Set Up WebSockets for Real-time Notifications

Install Django Channels:

BASHCopied
pip install channels channels-redis

Update settings.py:

PYTHONCopied
# settings.py
INSTALLED_APPS = [
    # ... other apps
    'channels',
]

ASGI_APPLICATION = 'myproject.asgi.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

Create a consumer for WebSocket connections:

PYTHONCopied
# yourapp/consumers.py
import json
from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_sync

class NotificationConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()
        
        # Join the notifications group
        async_to_sync(self.channel_layer.group_add)(
            "notifications",
            self.channel_name
        )
    
    def disconnect(self, close_code):
        # Leave the notifications group
        async_to_sync(self.channel_layer.group_discard)(
            "notifications",
            self.channel_name
        )
    
    def send_notification(self, event):
        # Send message to WebSocket
        self.send(text_data=json.dumps({
            'message': event['message']
        }))

Set up routing:

PYTHONCopied
# yourapp/routing.py
from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/notifications/$', consumers.NotificationConsumer.as_connect),
]

PYTHONCopied
# myproject/asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import yourapp.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            yourapp.routing.websocket_urlpatterns
        )
    ),
})

5. Frontend Implementation

Add this JavaScript to your template to handle notifications and play a sound:

HTMLCopied
<!-- In your template -->
<audio id="notification-sound" src="{% static 'sounds/notification.mp3' %}" preload="auto"></audio>

<script>
    const notificationSocket = new WebSocket(
        'ws://' + window.location.host + '/ws/notifications/'
    );

    notificationSocket.onmessage = function(e) {
        const data = JSON.parse(e.data);
        console.log('Notification received:', data.message);
        
        // Play sound
        document.getElementById('notification-sound').play();
        
        // Show notification
        alert(data.message);
    };

    notificationSocket.onclose = function(e) {
        console.error('Notification socket closed unexpectedly');
    };
</script>

Make sure to place a sound file at static/sounds/notification.mp3 or adjust the path accordingly.

Running the System

1. Start Redis: redis-server
2. Start Celery worker: celery -A myproject worker -l info
3. Start Celery beat: celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
4. Run your Django server: python manage.py runserver

This setup will check daily at 11:00 AM if a record has been created that day, and if not, it will send a real-time notification to your browser with a sound alert.