Dynamic blocks for Wagtail StreamFields

Say you're building a reusable Wagtail library that provides page models with StreamFields. You want users of your library to register their own blocks, maybe via Wagtail hooks.

The problem: every time a block is added, Django generates a migration. And that migration ends up in your library's code, not the user's project.

This post shows how to fix that with a custom DynamicStreamField.

Warning

Dynamic blocks are not officially supported. A Wagtail maintainer called it a "hack":

The Django ORM is fundamentally built on the premise that field definitions are fixed at the class level... making it dynamic is breaking the contract we've made with Django, and can only ever be a hack.

It works fine in practice. But know that you're on your own.

What We're Building

A block registry using Wagtail hooks:

# User's wagtail_hooks.py
from wagtail import hooks

@hooks.register("register_streamfield_blocks")
def register_my_blocks():
    from myblocks import QuoteBlock, CalloutBlock
    return [
        ("quote", QuoteBlock()),
        ("callout", CalloutBlock()),
    ]

These blocks show up in your StreamField. No migrations in the library.

The Registry

# registry.py
from wagtail import hooks

_block_cache = None

def get_registered_blocks():
    global _block_cache

    if _block_cache is not None:
        return _block_cache

    blocks = []
    for fn in hooks.get_hooks("register_streamfield_blocks"):
        result = fn()
        if result:
            blocks.extend(result)

    _block_cache = blocks
    return blocks

# blocks.py
from wagtail import blocks
from wagtail.images.blocks import ImageChooserBlock
from .registry import get_registered_blocks

BASE_BLOCKS = [
    ("heading", blocks.CharBlock(label="Heading", icon="title")),
    ("paragraph", blocks.RichTextBlock(label="Paragraph")),
    ("image", ImageChooserBlock(label="Image")),
]

def get_all_blocks():
    return BASE_BLOCKS + get_registered_blocks()

The Problem

Use it in a model:

from wagtail.models import Page
from wagtail.fields import StreamField
from .blocks import get_all_blocks

class ArticlePage(Page):
    body = StreamField(get_all_blocks(), blank=True)

Run makemigrations. Django bakes all blocks into the migration:

('body', wagtail.fields.StreamField(
    [('heading', 0), ('paragraph', 1), ('image', 2), ('quote', 3)],
    block_lookup={
        0: ('wagtail.blocks.CharBlock', (), {'label': 'Heading', 'icon': 'title'}),
        1: ('wagtail.blocks.RichTextBlock', (), {'label': 'Paragraph'}),
        2: ('wagtail.images.blocks.ImageChooserBlock', (), {'label': 'Image'}),
        3: ('wagtail.blocks.StructBlock', [[('quote', ...)]], {}),
    },
    blank=True
)),

Now a user adds a block via hooks:

$ python manage.py makemigrations --dry-run
Migrations for 'myapp':
  ~ Alter field body on articlepage

New migration. In your library. That's the problem.

This is a known issue. Some teams report migrations of tens of megabytes.

The Solution

A custom field that hides block info from migrations. Similar to what Mozilla Springfield and Cynthia Kiser did.

# fields.py
import json
from django.utils.functional import cached_property
from wagtail.blocks import StreamValue
from wagtail.fields import StreamField


class DynamicStreamField(StreamField):
    def __init__(
        self, block_types=None, use_json_field=True, block_lookup=None, **kwargs
    ):
        if block_types is None:
            block_types = []
        super().__init__(block_types, use_json_field, block_lookup, **kwargs)

    @cached_property
    def stream_block(self):
        if callable(self.block_types_arg):
            self.block_types_arg = self.block_types_arg()
        return super().stream_block

    def deconstruct(self):
        name, path, args, kwargs = super().deconstruct()
        kwargs["block_lookup"] = {}
        return name, path, [], kwargs

    def to_python(self, value):
        result = super().to_python(value)

        # In migrations, child_blocks is empty (block_lookup={}). 
        # Detect this and preserve raw data instead of losing it.
        if not self.stream_block.child_blocks and value and not result._raw_data:
            if isinstance(value, list):
                return StreamValue(self.stream_block, value, is_lazy=True)
            if isinstance(value, str):
                try:
                    parsed = json.loads(value)
                    return StreamValue(self.stream_block, parsed, is_lazy=True)
                except (ValueError, TypeError):
                    pass

        return result

How It Works

deconstruct() controls what Django stores in migrations and uses for comparison. We return empty args and block_lookup={}. Django sees no changes, makes no migrations.

stream_block runs the callable at runtime. Blocks work normally in your app.

to_python() prevents data loss during migrations. When Django runs a data migration, it reconstructs models from migration state. With block_lookup={}, the field has no child_blocks. The default StreamBlock.to_python() filters out blocks not in child_blocks - which means all your data gets filtered out. If your migration then calls page.save(), you lose everything.

This override detects the migration context (no child_blocks) and preserves the raw JSON instead of filtering it.

Usage

from .fields import DynamicStreamField
from .blocks import get_all_blocks

class ArticlePage(Page):
    body = DynamicStreamField(get_all_blocks, blank=True)

Both work:

• DynamicStreamField(get_all_blocks, ...) - callable
• DynamicStreamField(get_all_blocks(), ...) - executed

The migration now looks like:

('body', myapp.fields.DynamicStreamField(blank=True, block_lookup={})),

Add blocks, remove blocks, change blocks:

$ python manage.py makemigrations --dry-run
No changes detected

Data Migrations

What about StreamField data migrations? Like renaming a block?

Wagtail provides MigrateStreamData for this, but it doesn't work reliably with DynamicStreamField. The problem: migrations store block_lookup={}, so there are no block definitions to navigate the structure.

Top-Level Works

Top-level renames happen to work:

from wagtail.blocks.migrations.migrate_operation import MigrateStreamData
from wagtail.blocks.migrations.operations import RenameStreamChildrenOperation

class Migration(migrations.Migration):
    operations = [
        MigrateStreamData(
            app_name="myapp",
            model_name="ArticlePage",
            field_name="body",
            operations_and_block_paths=[
                (RenameStreamChildrenOperation(old_name="heading", new_name="title"), ""),
            ],
        ),
    ]

Why? For block_path="", Wagtail operates directly on the raw JSON without needing block definitions:

def apply(self, block_value):
    # Just dict operations on [{"type": "heading", "value": "Hello"}, ...]
    mapped_block_value = []
    for child_block in block_value:
        if child_block["type"] == self.old_name:
            mapped_block_value.append({**child_block, "type": self.new_name})
        else:
            mapped_block_value.append(child_block)
    return mapped_block_value

Nested Fails

But renaming blocks inside other blocks fails:

MigrateStreamData(
    app_name="myapp",
    model_name="ArticlePage", 
    field_name="body",
    operations_and_block_paths=[
        (RenameStreamChildrenOperation(old_name="text", new_name="paragraph"), "section.content"),
    ],
)

Error:

KeyError: 'section'
InvalidBlockDefError: No current block def named section

To navigate to section.content, Wagtail needs child_blocks. But child_blocks is empty because block_lookup={}.

Use Manual JSON Instead

Don't rely on MigrateStreamData. Write your data migrations with RunPython or a management command:

import json
from django.db import migrations

def rename_nested_blocks(apps, schema_editor):
    ArticlePage = apps.get_model('myapp', 'ArticlePage')

    for page in ArticlePage.objects.all():
        if not page.body:
            continue

        data = json.loads(page.body) if isinstance(page.body, str) else page.body
        modified = False

        for block in data:
            if block['type'] == 'section':
                for child in block['value'].get('content', []):
                    if child['type'] == 'text':
                        child['type'] = 'paragraph'
                        modified = True

        if modified:
            page.body = json.dumps(data)
            page.save(update_fields=['body'])

class Migration(migrations.Migration):
    operations = [
        migrations.RunPython(rename_nested_blocks, migrations.RunPython.noop),
    ]

No Wagtail machinery. Just Python dicts. Works for any block structure.

Full Code

registry.py

from wagtail import hooks

_block_cache = None

def get_registered_blocks():
    global _block_cache
    if _block_cache is not None:
        return _block_cache

    blocks = []
    for fn in hooks.get_hooks("register_streamfield_blocks"):
        result = fn()
        if result:
            blocks.extend(result)

    _block_cache = blocks
    return blocks

blocks.py

from wagtail import blocks
from wagtail.images.blocks import ImageChooserBlock
from .registry import get_registered_blocks

BASE_BLOCKS = [
    ("heading", blocks.CharBlock(label="Heading", icon="title")),
    ("paragraph", blocks.RichTextBlock(label="Paragraph")),
    ("image", ImageChooserBlock(label="Image")),
]

def get_all_blocks():
    return BASE_BLOCKS + get_registered_blocks()

fields.py

import json
from django.utils.functional import cached_property
from wagtail.blocks import StreamValue
from wagtail.fields import StreamField

class DynamicStreamField(StreamField):
    def __init__(self, block_types=None, use_json_field=True, block_lookup=None, **kwargs):
        if block_types is None:
            block_types = []
        super().__init__(block_types, use_json_field, block_lookup, **kwargs)

    @cached_property
    def stream_block(self):
        if callable(self.block_types_arg):
            self.block_types_arg = self.block_types_arg()
        return super().stream_block

    def deconstruct(self):
        name, path, args, kwargs = super().deconstruct()
        kwargs["block_lookup"] = {}
        return name, path, [], kwargs

    def to_python(self, value):
        result = super().to_python(value)

        # In migrations, child_blocks is empty (block_lookup={}). 
        # Detect this and preserve raw data instead of losing it.
        if not self.stream_block.child_blocks and value and not result._raw_data:
            if isinstance(value, list):
                return StreamValue(self.stream_block, value, is_lazy=True)
            if isinstance(value, str):
                try:
                    parsed = json.loads(value)
                    return StreamValue(self.stream_block, parsed, is_lazy=True)
                except (ValueError, TypeError):
                    pass

        return result

models.py

from wagtail.models import Page
from wagtail.admin.panels import FieldPanel
from .fields import DynamicStreamField
from .blocks import get_all_blocks

class ArticlePage(Page):
    body = DynamicStreamField(get_all_blocks, blank=True)

    content_panels = Page.content_panels + [
        FieldPanel("body"),
    ]

wagtail_hooks.py (user's project)

from wagtail import hooks

@hooks.register("register_streamfield_blocks")
def register_custom_blocks():
    from wagtail.blocks import CharBlock, TextBlock
    return [
        ("subtitle", CharBlock(label="Subtitle")),
        ("pullquote", TextBlock(label="Pull Quote")),
    ]