Python decorator and manager replacing recursive execution into sequential. Sometimes useful - my use case was with nested Django Rest Framework serializers, where updates had to happen in "layers".

sequential_execution.py

from contextlib import contextmanager
from contextvars import ContextVar
from functools import wraps, partial


_sequential_execution_active = ContextVar('bulk_update_active', False)
_sequential_execution_callbacks = ContextVar('bulk_update_callbacks', None)


@contextmanager
def sequential_execution():
    """
    Manager delaying execution of callbacks added by 'execute_sequentially' or 'sequential' decorator.
    Callbacks are executed when exiting manager. If new callbacks will be added during callback execution,
    they'll be executed as well.

    Used to replace recursive execution by sequential one. Lets consider these recursive calls:
    update('a') -> update('a.b') -> update('a.b.c')
                                 -> update('a.b.d')
                -> update('a.e') -> update('a.e.f')

    with recursive execution order of updates will be depth-first:
    'a' -> 'a.b' -> 'a.b.c' -> 'a.b.d' -> 'a.e' -> 'a.e.f.'

    if update function would be marked by sequential decorator and executed in sequential_execution context,
    order will be breath-first:
    'a' -> 'a.b' -> 'a.e' -> 'a.b.c' -> 'a.b.d' -> 'a.e.f.'
    """
    prev_value = _sequential_execution_active.get()
    try:
        _sequential_execution_active.set(True)
        yield
    finally:
        if prev_value is False:
            while True:
                callbacks = _sequential_execution_callbacks.get([])
                # resetting to the new one, so currently processed ones won't be modified
                _sequential_execution_callbacks.set([])
                if not callbacks:
                    break
                for callback in callbacks:
                    callback()
        _sequential_execution_active.set(prev_value)


def execute_sequentially(fn, *args, **kwargs):
    """Delays execution of fn if sequential_execution context is active. Otherwise, execute immediately."""
    if _sequential_execution_active.get():
        handler = partial(fn, *args, **kwargs)
        callbacks = _sequential_execution_callbacks.get([])
        callbacks.append(handler)
        _sequential_execution_callbacks.set(callbacks)
    else:
        return fn(*args, **kwargs)


def sequential(fn):
    """Decorator to automatically delay all calls of a given function"""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        execute_sequentially(fn, *args, **kwargs)
    return wrapper