You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
134 lines
5.5 KiB
134 lines
5.5 KiB
from django.core.management.base import BaseCommand
|
|
from django.db import transaction, connection
|
|
from django.db.models import ProtectedError
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from apps.course.models import (
|
|
Course, CourseCategory,
|
|
Lesson, CourseLesson, LessonCompletion,
|
|
Attachment, CourseAttachment,
|
|
Glossary, CourseGlossary,
|
|
Participant
|
|
)
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = _('Clear all course-related data from the database')
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
'--force',
|
|
action='store_true',
|
|
dest='force',
|
|
help=_('Force deletion without confirmation'),
|
|
)
|
|
parser.add_argument(
|
|
'--model',
|
|
type=str,
|
|
dest='model',
|
|
help=_('Specify a single model to clear (e.g., "Course", "Lesson", etc.)'),
|
|
)
|
|
parser.add_argument(
|
|
'--legacy-only',
|
|
action='store_true',
|
|
dest='legacy_only',
|
|
help=_('Clear only legacy models (before migration to new structure)'),
|
|
)
|
|
|
|
def table_exists(self, table_name):
|
|
"""Check if a table exists in the database."""
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT EXISTS (
|
|
SELECT FROM information_schema.tables
|
|
WHERE table_schema = 'public'
|
|
AND table_name = %s
|
|
);
|
|
""",
|
|
[table_name]
|
|
)
|
|
return cursor.fetchone()[0]
|
|
|
|
def handle(self, *args, **options):
|
|
force = options['force']
|
|
specific_model = options.get('model')
|
|
legacy_only = options.get('legacy_only')
|
|
|
|
if not force and not specific_model:
|
|
confirm = input(_('This will delete ALL course-related data. Are you sure? (yes/no): '))
|
|
if confirm.lower() != 'yes':
|
|
self.stdout.write(self.style.WARNING(_('Operation cancelled.')))
|
|
return
|
|
|
|
# Define all models
|
|
all_models = {
|
|
'Course': (Course, 'course_course'),
|
|
'CourseCategory': (CourseCategory, 'course_coursecategory'),
|
|
'Lesson': (Lesson, 'course_lesson'),
|
|
'CourseLesson': (CourseLesson, 'course_courselesson'),
|
|
'LessonCompletion': (LessonCompletion, 'course_lessoncompletion'),
|
|
'Attachment': (Attachment, 'course_attachment'),
|
|
'CourseAttachment': (CourseAttachment, 'course_courseattachment'),
|
|
'Glossary': (Glossary, 'course_glossary'),
|
|
'CourseGlossary': (CourseGlossary, 'course_courseglossary'),
|
|
'Participant': (Participant, 'course_participant'),
|
|
}
|
|
|
|
# Legacy models (before migration)
|
|
legacy_models = {
|
|
'Course': (Course, 'course_course'),
|
|
'CourseCategory': (CourseCategory, 'course_coursecategory'),
|
|
'Lesson': (Lesson, 'course_lesson'),
|
|
'LessonCompletion': (LessonCompletion, 'course_lessoncompletion'),
|
|
'Attachment': (Attachment, 'course_attachment'),
|
|
'Glossary': (Glossary, 'course_glossary'),
|
|
'Participant': (Participant, 'course_participant'),
|
|
}
|
|
|
|
models_to_use = legacy_models if legacy_only else all_models
|
|
|
|
if specific_model:
|
|
# Clear only the specified model
|
|
if specific_model not in models_to_use:
|
|
self.stdout.write(self.style.ERROR(_(f'Unknown model: {specific_model}')))
|
|
self.stdout.write(self.style.WARNING(_(f'Available models: {", ".join(models_to_use.keys())}')))
|
|
return
|
|
|
|
model_info = models_to_use[specific_model]
|
|
models_to_clear = [(specific_model, model_info[0], model_info[1])]
|
|
else:
|
|
# Clear all models in the correct order to avoid foreign key constraints
|
|
models_to_clear = []
|
|
|
|
# Order matters for foreign key constraints
|
|
model_order = [
|
|
'LessonCompletion', 'CourseLesson', 'Lesson',
|
|
'CourseAttachment', 'Attachment',
|
|
'CourseGlossary', 'Glossary',
|
|
'Participant', 'Course', 'CourseCategory'
|
|
]
|
|
|
|
for model_name in model_order:
|
|
if model_name in models_to_use:
|
|
model_info = models_to_use[model_name]
|
|
models_to_clear.append((model_name, model_info[0], model_info[1]))
|
|
|
|
# Process each model
|
|
for model_name, model_class, table_name in models_to_clear:
|
|
# Check if the table exists
|
|
if not self.table_exists(table_name):
|
|
self.stdout.write(self.style.WARNING(_(f'Table {table_name} does not exist, skipping {model_name}')))
|
|
continue
|
|
|
|
try:
|
|
count = model_class.objects.count()
|
|
model_class.objects.all().delete()
|
|
self.stdout.write(self.style.SUCCESS(_(f'Deleted {count} {model_name} records')))
|
|
except ProtectedError as e:
|
|
self.stdout.write(self.style.ERROR(_(f'Could not delete {model_name} records due to protected foreign keys')))
|
|
self.stdout.write(self.style.ERROR(str(e)))
|
|
except Exception as e:
|
|
self.stdout.write(self.style.ERROR(_(f'Error deleting {model_name} records: {str(e)}')))
|
|
|
|
self.stdout.write(self.style.SUCCESS(_('Course data clearing completed')))
|