openedx-authz/openedx_authz/management/commands/authz_migrate_course_authoring.py at 0184ddd5f7b39a89e7f868d070a2d951ae952a74 · openedx/openedx-authz · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
"""
Django management command to migrate legacy course authoring roles to the new Authz (Casbin-based) authorization system.
"""

from django.core.management.base import BaseCommand, CommandError
from django.db import transaction

from openedx_authz.engine.utils import migrate_legacy_course_roles_to_authz

try:
    from common.djangoapps.student.models import CourseAccessRole
except ImportError:
    CourseAccessRole = None  # type: ignore


class Command(BaseCommand):
    """
    Django command to migrate legacy CourseAccessRole data
    to the new Authz (Casbin-based) authorization system.
    """

    help = "Migrate legacy course authoring roles to the new Authz system."

    def add_arguments(self, parser):
        parser.add_argument(
            "--delete",
            action="store_true",
            help="Delete legacy CourseAccessRole records after successful migration.",
        )
        parser.add_argument(
            "--course-id-list",
            nargs="+",
            type=str,
            help="Optional list of course IDs to filter the migration.",
        )

        parser.add_argument(
            "--org-id",
            type=str,
            help="Optional organization ID to filter the migration.",
        )

    def handle(self, *args, **options):
        delete_after_migration = options["delete"]
        course_id_list = options.get("course_id_list")
        org_id = options.get("org_id")

        if not course_id_list and not org_id:
            raise CommandError("You must specify either --course-id-list or --org-id to filter the migration.")

        if course_id_list and org_id:
            raise CommandError("You cannot use --course-id-list and --org-id together.")

        self.stdout.write(self.style.WARNING("Starting legacy → Authz migration..."))

        try:
            if delete_after_migration:
                confirm = input(
                    "Are you sure you want to delete successfully migrated legacy roles? Type 'yes' to continue: "
                )

                if confirm != "yes":
                    self.stdout.write(self.style.WARNING("Migration aborted."))
                    return
            with transaction.atomic():
                errors, success = migrate_legacy_course_roles_to_authz(
                    course_access_role_model=CourseAccessRole,
                    course_id_list=course_id_list,
                    org_id=org_id,
                    delete_after_migration=delete_after_migration,
                )

                if errors and success:
                    self.stdout.write(
                        self.style.WARNING(
                            f"Migration completed with {len(errors)} errors and {len(success)} roles migrated."
                        )
                    )
                elif errors:
                    self.stdout.write(self.style.ERROR(f"Migration completed with {len(errors)} errors."))
                elif success:
                    self.stdout.write(
                        self.style.SUCCESS(f"Migration completed successfully with {len(success)} roles migrated.")
                    )
                else:
                    self.stdout.write(
                        self.style.ERROR(
                            "No legacy roles found for the given scope, course could already be migrated, "
                            "or there could be an error in the course_id_list / org_id."
                        )
                    )

                if delete_after_migration:
                    self.stdout.write(self.style.SUCCESS(f"{len(success)} Legacy roles deleted successfully."))

        except Exception as exc:
            self.stdout.write(self.style.ERROR(f"Migration failed due to unexpected error: {exc}"))
            raise

        self.stdout.write(self.style.SUCCESS("Done."))