StreamField 마이그레이션¶

import json

from django.core.serializers.json import DjangoJSONEncoder
from django.db import migrations

import wagtail.blocks
import wagtail.fields


def convert_to_streamfield(apps, schema_editor):
    BlogPage = apps.get_model("demo", "BlogPage")
    for page in BlogPage.objects.all():
        page.body = json.dumps(
            [{"type": "rich_text", "value": page.body}],
            cls=DjangoJSONEncoder
        )
        page.save()


def convert_to_richtext(apps, schema_editor):
    BlogPage = apps.get_model("demo", "BlogPage")
    for page in BlogPage.objects.all():
        if page.body:
            stream = json.loads(page.body)
            page.body = "".join([
                child["value"] for child in stream
                if child["type"] == "rich_text"
            ])
            page.save()


class Migration(migrations.Migration):

    dependencies = [
        # 생성된 마이그레이션의 의존성(dependency) 라인을 그대로 두십시오!
        ("demo", "0001_initial"),
    ]

    operations = [
        migrations.RunPython(
            convert_to_streamfield,
            convert_to_richtext,
        ),

        # 생성된 AlterField를 그대로 두십시오!
        migrations.AlterField(
            model_name="BlogPage",
            name="body",
            field=wagtail.fields.StreamField(
                [("rich_text", wagtail.blocks.RichTextBlock())],
            ),
        ),
    ]

import json

from django.contrib.contenttypes.models import ContentType
from django.core.serializers.json import DjangoJSONEncoder
from django.db import migrations

import wagtail.blocks
import wagtail.fields


def page_to_streamfield(page):
    changed = False
    try:
        json.loads(page.body)
    except ValueError:
        page.body = json.dumps(
            [{"type": "rich_text", "value": page.body}],
        )
        changed = True
    else:
        # 이미 유효한 JSON입니다. 그대로 두십시오.
        pass

    return page, changed


def pagerevision_to_streamfield(revision_data):
    changed = False
    body = revision_data.get("body")
    if body:
        try:
            json.loads(body)
        except ValueError:
            revision_data["body"] = json.dumps(
                [{
                    "value": body,
                    "type": "rich_text"
                }],
                cls=DjangoJSONEncoder)
            changed = True
        else:
            # 이미 유효한 JSON입니다. 그대로 두십시오.
            pass
    return revision_data, changed


def page_to_richtext(page):
    changed = False
    if page.body:
        try:
            body_data = json.loads(page.body)
        except ValueError:
            # StreamField가 아닌 것 같습니다. 그대로 두십시오.
            pass
        else:
            page.body = "".join([
                child["value"] for child in body_data
                if child["type"] == "rich_text"
            ])
            changed = True

    return page, changed


def pagerevision_to_richtext(revision_data):
    changed = False
    body = revision_data.get("body", "definitely non-JSON string")
    if body:
        try:
            body_data = json.loads(body)
        except ValueError:
            # StreamField가 아닌 것 같습니다. 그대로 두십시오.
            pass
        else:
            raw_text = "".join([
                child["value"] for child in body_data
                if child["type"] == "rich_text"
            ])
            revision_data["body"] = raw_text
            changed = True
    return revision_data, changed


def convert(apps, schema_editor, page_converter, pagerevision_converter):
    BlogPage = apps.get_model("demo", "BlogPage")
    content_type = ContentType.objects.get_for_model(BlogPage)
    Revision = apps.get_model("wagtailcore", "Revision")

    for page in BlogPage.objects.all():

        page, changed = page_converter(page)
        if changed:
            page.save()

        for revision in Revision.objects.filter(
            content_type_id=content_type.pk, object_id=page.pk
        ):
            revision_data = revision.content
            revision_data, changed = pagerevision_converter(revision_data)
            if changed:
                revision.content = revision_data
                revision.save()


def convert_to_streamfield(apps, schema_editor):
    return convert(apps, schema_editor, page_to_streamfield, pagerevision_to_streamfield)


def convert_to_richtext(apps, schema_editor):
    return convert(apps, schema_editor, page_to_richtext, pagerevision_to_richtext)


class Migration(migrations.Migration):

    dependencies = [
        # 생성된 마이그레이션의 의존성 라인을 그대로 두십시오!
        ("demo", "0001_initial"),
        ("wagtailcore", "0076_modellogentry_revision"),
    ]

    operations = [
        migrations.RunPython(
            convert_to_streamfield,
            convert_to_richtext,
        ),

        # 생성된 AlterField를 그대로 두십시오!
        migrations.AlterField(
            model_name="BlogPage",
            name="body",
            field=wagtail.fields.StreamField(
                [("rich_text", wagtail.blocks.RichTextBlock())],
            ),
        ),
    ]