router

Tier 2 convenience router with auto-wired reorder, remove, and clear endpoints.

init_sortable_queue_router


def init_sortable_queue_router(
    config:SortableQueueConfig, # Queue configuration
    get_items:Callable, # (session_id) -> current items
    set_items:Callable, # (session_id, items) -> persist
    render_content:Callable, # Custom content callback
    on_mutate:Optional=None, # (mutation_type, items, sess) -> OOB elements
    prefix:str='/queue', # Route prefix
    render_kwargs:VAR_KEYWORD
)->Tuple: # (router, urls) tuple

Initialize a router with reorder, remove, and clear endpoints.

The on_mutate callback receives the mutation type (“reorder”, “remove”, “clear”), the updated items list, and the session ID. It should return a tuple of OOB elements to append to the response, or an empty tuple.

Tests

from fasthtml.common import Span

# Test setup
config = SortableQueueConfig(prefix="rt")
state = {"items": [{"id": "a"}, {"id": "b"}]}

def get_items(sess):
    return list(state["items"])

def set_items(sess, items):
    state["items"] = items

def content_fn(item, index):
    return Span(item["id"])

# --- Router creation ---
router, urls = init_sortable_queue_router(
    config=config,
    get_items=get_items,
    set_items=set_items,
    render_content=content_fn,
    prefix="/test-queue",
)

# Verify return types
assert isinstance(router, APIRouter)
assert isinstance(urls, SortableQueueUrls)

# Verify URL paths
assert urls.reorder == "/test-queue/reorder"
assert urls.remove == "/test-queue/remove"
assert urls.clear == "/test-queue/clear"

# --- With on_mutate callback ---
mutate_log = []

def on_mutate(mutation_type, items, sess):
    mutate_log.append((mutation_type, len(items)))
    return ()

router2, urls2 = init_sortable_queue_router(
    config=config,
    get_items=get_items,
    set_items=set_items,
    render_content=content_fn,
    on_mutate=on_mutate,
    prefix="/q2",
)
assert urls2.reorder == "/q2/reorder"

print("All router tests passed")

All router tests passed