Finding Blocks Across a Wagtail Site

wagtail_find_block.py

from wagtail.blocks import StreamValue, StructValue
from wagtail.blocks.list_block import ListValue
from wagtail.models import Page
from wagtail.fields import StreamField


def find_block(bound_block, *criterias, parent_bound_block=None):
    if all(criteria(bound_block) for criteria in criterias):
        yield (parent_bound_block, bound_block)
    else:
        value = bound_block.value
        if isinstance(value, StreamValue):
            for child in value:
                yield from find_block(child, *criterias, parent_bound_block=bound_block)
        elif isinstance(value, ListValue):
            for child in value.bound_blocks:
                yield from find_block(child, *criterias, parent_bound_block=bound_block)
        elif isinstance(value, StructValue):
            for child in value.bound_blocks.values():
                yield from find_block(child, *criterias, parent_bound_block=bound_block)


def find_block_in_page(page, *criterias):
    for field in page._meta.get_fields():
        if isinstance(field, StreamField):
            for index, block in enumerate(getattr(page, field.name)):
                for parent_bound_block, bound_block in find_block(block, *criterias):
                    yield (field, index, parent_bound_block, bound_block)


def find_block_in_pages(*criterias, qs=None):
    if qs is None:
        qs = Page.objects.all()

    for page in qs.specific():
        for field, index, parent_bound_block, bound_block in find_block_in_page(
            page, *criterias
        ):
            yield (page, field, index, parent_bound_block, bound_block)


def block_instance_of(block_class):
    def criteria(bound_block):
        return isinstance(bound_block.block, block_class)

    return criteria