Notes

# admin.py
from django.contrib import admin
from .models import Employee
from .tasks import generate_csv_data

@admin.register(Employee)
class EmployeeAdmin(admin.ModelAdmin):
list_display = ('name', 'email', 'company_name', 'employee_id', 'created_at')

def get_urls(self):
from django.urls import path
urls = super().get_urls()
custom_urls = [
path(
'generate-data/',
self.admin_site.admin_view(self.generate_data_view),
name='generate-data',
),
]
return custom_urls + urls

def generate_data_view(self, request):
from django.shortcuts import render
from django.http import JsonResponse
if request.method == "POST":
num_records = int(request.POST.get("num_records"))
generate_csv_data.delay(num_records)
return JsonResponse({"message": f"Task to generate {num_records} records started!"})
return render(request, 'admin/generate_data.html')

# tasks.py
import pandas as pd
from faker import Faker
from celery import shared_task
from django.conf import settings
import os

@shared_task
def generate_csv_data(num_records):
fake = Faker()
data = [
{
"name": fake.name(),
"email": fake.email(),
"company_name": fake.company(),
"employee_id": fake.uuid4()[:8], # Generate unique ID
}
for _ in range(num_records)
]

# Save to CSV
df = pd.DataFrame(data)
csv_path = os.path.join(settings.MEDIA_ROOT, 'generated_data.csv')
df.to_csv(csv_path, index=False)

# Trigger import task after delay
import_csv_data.delay(csv_path)
return csv_path

@shared_task
def import_csv_data(csv_path):
from .models import Employee
df = pd.read_csv(csv_path)

for _, row in df.iterrows():
Employee.objects.create(
name=row['name'],
email=row['email'],
company_name=row['company_name'],
employee_id=row['employee_id']
)
return f"Imported data from {csv_path}"


# tasks.py (continuation)
from django.core.mail import send_mail

@shared_task
def send_reminder_email(user_email, record_count, timestamp):
subject = "Data Import Reminder"
message = f"{record_count} records were imported into the system on {timestamp}."
send_mail(subject, message, '[email protected]', [user_email])


# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

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

app = Celery('your_project')

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

app.autodiscover_tasks()


<!-- templates/admin/generate_data.html -->
{% extends "admin/base_site.html" %}

{% block content %}
<h1>Generate Fake Data</h1>
<form method="post">
{% csrf_token %}
<label for="num_records">Number of Records:</label>
<input type="number" id="num_records" name="num_records" required>
<button type="submit">Generate</button>
</form>
{% endblock %}



# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
MEDIA_URL = '/media/'

celery -A your_project worker --loglevel=info
celery -A your_project beat --loglevel=info