Implement isomorphic Phase 1: per-page component bundling
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 4m43s
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 4m43s
Add component dependency analyzer (shared/sx/deps.py) that walks component AST bodies to compute transitive dependency sets. sx_page() and sx_response() now send only the components each page needs instead of the entire registry. Changes: - New: shared/sx/deps.py — transitive_deps(), components_needed(), scan_components_from_sx(), compute_all_deps() - shared/sx/types.py — Add deps: set[str] field to Component - shared/sx/jinja_bridge.py — Compute deps on registration, add components_for_page() and css_classes_for_page() - shared/sx/helpers.py — sx_page() uses per-page bundle + hash, sx_response() passes source to components_for_request() for page-scoped component diffing - New: shared/sx/tests/test_deps.py — 15 tests covering AST scanning, transitive deps, circular refs, per-page bundling Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
114
shared/sx/deps.py
Normal file
114
shared/sx/deps.py
Normal file
@@ -0,0 +1,114 @@
|
||||
"""
|
||||
Component dependency analysis.
|
||||
|
||||
Walks component AST bodies to compute transitive dependency sets.
|
||||
A component's deps are all other components (~name references) it
|
||||
can potentially render, including through control flow branches.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .types import Component, Macro, Symbol
|
||||
|
||||
|
||||
def _scan_ast(node: Any) -> set[str]:
|
||||
"""Scan an AST node for ~component references.
|
||||
|
||||
Walks all branches of control flow (if/when/cond/case) to find
|
||||
every component that *could* be rendered. Returns a set of
|
||||
component names (with ~ prefix).
|
||||
"""
|
||||
refs: set[str] = set()
|
||||
_walk(node, refs)
|
||||
return refs
|
||||
|
||||
|
||||
def _walk(node: Any, refs: set[str]) -> None:
|
||||
"""Recursively walk an AST node collecting ~name references."""
|
||||
if isinstance(node, Symbol):
|
||||
if node.name.startswith("~"):
|
||||
refs.add(node.name)
|
||||
return
|
||||
|
||||
if isinstance(node, list):
|
||||
for item in node:
|
||||
_walk(item, refs)
|
||||
return
|
||||
|
||||
if isinstance(node, dict):
|
||||
for v in node.values():
|
||||
_walk(v, refs)
|
||||
return
|
||||
|
||||
# Literals (str, int, float, bool, None, Keyword) — no refs
|
||||
return
|
||||
|
||||
|
||||
def transitive_deps(name: str, env: dict[str, Any]) -> set[str]:
|
||||
"""Compute transitive component dependencies for *name*.
|
||||
|
||||
Returns the set of all component names (with ~ prefix) that
|
||||
*name* can transitively render, NOT including *name* itself.
|
||||
"""
|
||||
seen: set[str] = set()
|
||||
|
||||
def walk(n: str) -> None:
|
||||
if n in seen:
|
||||
return
|
||||
seen.add(n)
|
||||
val = env.get(n)
|
||||
if isinstance(val, Component):
|
||||
for dep in _scan_ast(val.body):
|
||||
walk(dep)
|
||||
elif isinstance(val, Macro):
|
||||
for dep in _scan_ast(val.body):
|
||||
walk(dep)
|
||||
|
||||
key = name if name.startswith("~") else f"~{name}"
|
||||
walk(key)
|
||||
return seen - {key}
|
||||
|
||||
|
||||
def compute_all_deps(env: dict[str, Any]) -> None:
|
||||
"""Compute and cache deps for all Component entries in *env*.
|
||||
|
||||
Mutates each Component's ``deps`` field in place.
|
||||
"""
|
||||
for key, val in env.items():
|
||||
if isinstance(val, Component):
|
||||
val.deps = transitive_deps(key, env)
|
||||
|
||||
|
||||
def scan_components_from_sx(source: str) -> set[str]:
|
||||
"""Extract component names referenced in SX source text.
|
||||
|
||||
Uses regex to find (~name patterns in serialized SX wire format.
|
||||
Returns names with ~ prefix, e.g. {"~card", "~nav-link"}.
|
||||
"""
|
||||
import re
|
||||
return set(re.findall(r'\(~([a-zA-Z_][a-zA-Z0-9_\-]*)', source))
|
||||
|
||||
|
||||
def components_needed(page_sx: str, env: dict[str, Any]) -> set[str]:
|
||||
"""Compute the full set of component names needed for a page.
|
||||
|
||||
Scans *page_sx* for direct component references, then computes
|
||||
the transitive closure over the component dependency graph.
|
||||
Returns names with ~ prefix.
|
||||
"""
|
||||
# Direct refs from the page source
|
||||
direct = {f"~{n}" for n in scan_components_from_sx(page_sx)}
|
||||
|
||||
# Transitive closure
|
||||
all_needed: set[str] = set()
|
||||
for name in direct:
|
||||
all_needed.add(name)
|
||||
val = env.get(name)
|
||||
if isinstance(val, Component) and val.deps:
|
||||
all_needed.update(val.deps)
|
||||
else:
|
||||
# deps not cached yet — compute on the fly
|
||||
all_needed.update(transitive_deps(name, env))
|
||||
|
||||
return all_needed
|
||||
@@ -456,34 +456,39 @@ def sx_call(component_name: str, **kwargs: Any) -> str:
|
||||
|
||||
|
||||
|
||||
def components_for_request() -> str:
|
||||
def components_for_request(source: str = "") -> str:
|
||||
"""Return defcomp/defmacro source for definitions the client doesn't have yet.
|
||||
|
||||
Reads the ``SX-Components`` header (comma-separated component names
|
||||
like ``~card,~nav-item``) and returns only the definitions the client
|
||||
is missing. If the header is absent, returns all defs.
|
||||
is missing. If *source* is provided, only sends components needed
|
||||
for that source (plus transitive deps). If the header is absent,
|
||||
returns all needed defs.
|
||||
"""
|
||||
from quart import request
|
||||
from .jinja_bridge import client_components_tag, _COMPONENT_ENV
|
||||
from .jinja_bridge import _COMPONENT_ENV
|
||||
from .deps import components_needed
|
||||
from .types import Component, Macro
|
||||
from .parser import serialize
|
||||
|
||||
loaded_raw = request.headers.get("SX-Components", "")
|
||||
if not loaded_raw:
|
||||
# Client has nothing — send all
|
||||
tag = client_components_tag()
|
||||
if not tag:
|
||||
return ""
|
||||
start = tag.find(">") + 1
|
||||
end = tag.rfind("</script>")
|
||||
return tag[start:end] if start > 0 and end > start else ""
|
||||
# Determine which components the page needs
|
||||
if source:
|
||||
needed = components_needed(source, _COMPONENT_ENV)
|
||||
else:
|
||||
needed = None # all
|
||||
|
||||
loaded_raw = request.headers.get("SX-Components", "")
|
||||
loaded = set(loaded_raw.split(",")) if loaded_raw else set()
|
||||
|
||||
loaded = set(loaded_raw.split(","))
|
||||
parts = []
|
||||
for key, val in _COMPONENT_ENV.items():
|
||||
if isinstance(val, Component):
|
||||
comp_name = f"~{val.name}"
|
||||
# Skip if not needed for this page
|
||||
if needed is not None and comp_name not in needed and key not in needed:
|
||||
continue
|
||||
# Skip components the client already has
|
||||
if f"~{val.name}" in loaded or val.name in loaded:
|
||||
if comp_name in loaded or val.name in loaded:
|
||||
continue
|
||||
# Reconstruct defcomp source
|
||||
param_strs = ["&key"] + list(val.params)
|
||||
@@ -530,7 +535,7 @@ def sx_response(source: str, status: int = 200,
|
||||
# For SX requests, prepend missing component definitions
|
||||
comp_defs = ""
|
||||
if request.headers.get("SX-Request"):
|
||||
comp_defs = components_for_request()
|
||||
comp_defs = components_for_request(source)
|
||||
if comp_defs:
|
||||
body = (f'<script type="text/sx" data-components>'
|
||||
f'{comp_defs}</script>\n{body}')
|
||||
@@ -641,11 +646,11 @@ def sx_page(ctx: dict, page_sx: str, *,
|
||||
renders everything client-side. CSS rules are scanned from the sx
|
||||
source and component defs, then injected as a <style> block.
|
||||
"""
|
||||
from .jinja_bridge import client_components_tag, _COMPONENT_ENV, get_component_hash
|
||||
from .css_registry import scan_classes_from_sx, lookup_rules, get_preamble, registry_loaded, store_css_hash
|
||||
from .types import Component
|
||||
from .jinja_bridge import components_for_page, css_classes_for_page
|
||||
from .css_registry import lookup_rules, get_preamble, registry_loaded, store_css_hash
|
||||
|
||||
component_hash = get_component_hash()
|
||||
# Per-page component bundle: only definitions this page needs
|
||||
component_defs, component_hash = components_for_page(page_sx)
|
||||
|
||||
# Check if client already has this version cached (via cookie)
|
||||
# In dev mode, always send full source so edits are visible immediately
|
||||
@@ -653,28 +658,13 @@ def sx_page(ctx: dict, page_sx: str, *,
|
||||
if not _is_dev_mode() and client_hash and client_hash == component_hash:
|
||||
# Client has current components cached — send empty source
|
||||
component_defs = ""
|
||||
else:
|
||||
components_tag = client_components_tag()
|
||||
# Extract just the inner source from the <script> tag
|
||||
component_defs = ""
|
||||
if components_tag:
|
||||
start = components_tag.find(">") + 1
|
||||
end = components_tag.rfind("</script>")
|
||||
if start > 0 and end > start:
|
||||
component_defs = components_tag[start:end]
|
||||
|
||||
# Scan for CSS classes — use pre-computed sets for components, scan page sx at request time
|
||||
# Scan for CSS classes — only from components this page uses + page source
|
||||
sx_css = ""
|
||||
sx_css_classes = ""
|
||||
sx_css_hash = ""
|
||||
if registry_loaded():
|
||||
# Union pre-computed component classes instead of re-scanning source
|
||||
classes: set[str] = set()
|
||||
for val in _COMPONENT_ENV.values():
|
||||
if isinstance(val, Component) and val.css_classes:
|
||||
classes.update(val.css_classes)
|
||||
# Page sx is unique per request — scan it
|
||||
classes.update(scan_classes_from_sx(page_sx))
|
||||
classes = css_classes_for_page(page_sx)
|
||||
# Always include body classes
|
||||
classes.update(["bg-stone-50", "text-stone-900"])
|
||||
rules = lookup_rules(classes)
|
||||
|
||||
@@ -198,6 +198,10 @@ def register_components(sx_source: str) -> None:
|
||||
all_classes = scan_classes_from_sx(sx_source)
|
||||
val.css_classes = set(all_classes)
|
||||
|
||||
# Recompute transitive deps for all components (cheap — just AST walking)
|
||||
from .deps import compute_all_deps
|
||||
compute_all_deps(_COMPONENT_ENV)
|
||||
|
||||
_compute_component_hash()
|
||||
|
||||
|
||||
@@ -322,6 +326,71 @@ def client_components_tag(*names: str) -> str:
|
||||
return f'<script type="text/sx" data-components>{source}</script>'
|
||||
|
||||
|
||||
def components_for_page(page_sx: str) -> tuple[str, str]:
|
||||
"""Return (component_defs_source, page_hash) for a page.
|
||||
|
||||
Scans *page_sx* for component references, computes the transitive
|
||||
closure, and returns only the definitions needed for this page.
|
||||
The hash is computed from the page-specific bundle for caching.
|
||||
"""
|
||||
from .deps import components_needed
|
||||
from .parser import serialize
|
||||
|
||||
needed = components_needed(page_sx, _COMPONENT_ENV)
|
||||
if not needed:
|
||||
return "", ""
|
||||
|
||||
# Also include macros — they're needed for client-side expansion
|
||||
parts = []
|
||||
for key, val in _COMPONENT_ENV.items():
|
||||
if isinstance(val, Component):
|
||||
if f"~{val.name}" in needed or key in needed:
|
||||
param_strs = ["&key"] + list(val.params)
|
||||
if val.has_children:
|
||||
param_strs.extend(["&rest", "children"])
|
||||
params_sx = "(" + " ".join(param_strs) + ")"
|
||||
body_sx = serialize(val.body, pretty=True)
|
||||
parts.append(f"(defcomp ~{val.name} {params_sx} {body_sx})")
|
||||
elif isinstance(val, Macro):
|
||||
# Include macros that are referenced in needed components' bodies
|
||||
# For now, include all macros (they're small and often shared)
|
||||
param_strs = list(val.params)
|
||||
if val.rest_param:
|
||||
param_strs.extend(["&rest", val.rest_param])
|
||||
params_sx = "(" + " ".join(param_strs) + ")"
|
||||
body_sx = serialize(val.body, pretty=True)
|
||||
parts.append(f"(defmacro {val.name} {params_sx} {body_sx})")
|
||||
|
||||
if not parts:
|
||||
return "", ""
|
||||
|
||||
source = "\n".join(parts)
|
||||
digest = hashlib.sha256(source.encode()).hexdigest()[:12]
|
||||
return source, digest
|
||||
|
||||
|
||||
def css_classes_for_page(page_sx: str) -> set[str]:
|
||||
"""Return CSS classes needed for a page's component bundle + page source.
|
||||
|
||||
Instead of unioning ALL component CSS classes, only includes classes
|
||||
from components the page actually uses.
|
||||
"""
|
||||
from .deps import components_needed
|
||||
from .css_registry import scan_classes_from_sx
|
||||
|
||||
needed = components_needed(page_sx, _COMPONENT_ENV)
|
||||
classes: set[str] = set()
|
||||
|
||||
for key, val in _COMPONENT_ENV.items():
|
||||
if isinstance(val, Component):
|
||||
if (f"~{val.name}" in needed or key in needed) and val.css_classes:
|
||||
classes.update(val.css_classes)
|
||||
|
||||
# Page sx is unique per request — scan it
|
||||
classes.update(scan_classes_from_sx(page_sx))
|
||||
return classes
|
||||
|
||||
|
||||
def sx_css_all() -> str:
|
||||
"""Return all CSS rules (preamble + utilities) for Jinja fallback pages."""
|
||||
from .css_registry import get_all_css
|
||||
|
||||
176
shared/sx/tests/test_deps.py
Normal file
176
shared/sx/tests/test_deps.py
Normal file
@@ -0,0 +1,176 @@
|
||||
"""Tests for the component dependency analyzer."""
|
||||
|
||||
import pytest
|
||||
from shared.sx.parser import parse_all
|
||||
from shared.sx.types import Component, Macro, Symbol
|
||||
from shared.sx.deps import (
|
||||
_scan_ast,
|
||||
transitive_deps,
|
||||
compute_all_deps,
|
||||
scan_components_from_sx,
|
||||
components_needed,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def make_env(*sx_sources: str) -> dict:
|
||||
"""Parse and evaluate component definitions into an env dict."""
|
||||
from shared.sx.evaluator import _eval, _trampoline
|
||||
env: dict = {}
|
||||
for source in sx_sources:
|
||||
exprs = parse_all(source)
|
||||
for expr in exprs:
|
||||
_trampoline(_eval(expr, env))
|
||||
return env
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _scan_ast
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestScanAst:
|
||||
def test_simple_component_ref(self):
|
||||
env = make_env('(defcomp ~card (&key title) (div (~badge :label title)))')
|
||||
comp = env["~card"]
|
||||
refs = _scan_ast(comp.body)
|
||||
assert refs == {"~badge"}
|
||||
|
||||
def test_no_refs(self):
|
||||
env = make_env('(defcomp ~plain (&key text) (div :class "p-4" text))')
|
||||
comp = env["~plain"]
|
||||
refs = _scan_ast(comp.body)
|
||||
assert refs == set()
|
||||
|
||||
def test_multiple_refs(self):
|
||||
env = make_env(
|
||||
'(defcomp ~page (&key title) (div (~header :title title) (~footer)))'
|
||||
)
|
||||
comp = env["~page"]
|
||||
refs = _scan_ast(comp.body)
|
||||
assert refs == {"~header", "~footer"}
|
||||
|
||||
def test_nested_in_control_flow(self):
|
||||
env = make_env(
|
||||
'(defcomp ~card (&key big) '
|
||||
' (if big (~big-card) (~small-card)))'
|
||||
)
|
||||
comp = env["~card"]
|
||||
refs = _scan_ast(comp.body)
|
||||
assert refs == {"~big-card", "~small-card"}
|
||||
|
||||
def test_refs_in_dict(self):
|
||||
env = make_env(
|
||||
'(defcomp ~wrap (&key) (div {:slot (~inner)}))'
|
||||
)
|
||||
comp = env["~wrap"]
|
||||
refs = _scan_ast(comp.body)
|
||||
assert refs == {"~inner"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# transitive_deps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestTransitiveDeps:
|
||||
def test_direct_dep(self):
|
||||
env = make_env(
|
||||
'(defcomp ~card (&key) (div (~badge)))',
|
||||
'(defcomp ~badge (&key) (span "★"))',
|
||||
)
|
||||
deps = transitive_deps("~card", env)
|
||||
assert deps == {"~badge"}
|
||||
|
||||
def test_transitive(self):
|
||||
env = make_env(
|
||||
'(defcomp ~page (&key) (div (~layout)))',
|
||||
'(defcomp ~layout (&key) (div (~header) (~footer)))',
|
||||
'(defcomp ~header (&key) (nav "header"))',
|
||||
'(defcomp ~footer (&key) (footer "footer"))',
|
||||
)
|
||||
deps = transitive_deps("~page", env)
|
||||
assert deps == {"~layout", "~header", "~footer"}
|
||||
|
||||
def test_circular(self):
|
||||
"""Circular deps should not cause infinite recursion."""
|
||||
env = make_env(
|
||||
'(defcomp ~a (&key) (div (~b)))',
|
||||
'(defcomp ~b (&key) (div (~a)))',
|
||||
)
|
||||
deps = transitive_deps("~a", env)
|
||||
assert deps == {"~b"}
|
||||
|
||||
def test_no_deps(self):
|
||||
env = make_env('(defcomp ~leaf (&key) (span "hi"))')
|
||||
deps = transitive_deps("~leaf", env)
|
||||
assert deps == set()
|
||||
|
||||
def test_missing_component(self):
|
||||
"""Referencing a component not in env should not crash."""
|
||||
env = make_env('(defcomp ~card (&key) (div (~unknown)))')
|
||||
deps = transitive_deps("~card", env)
|
||||
assert "~unknown" in deps
|
||||
|
||||
def test_without_tilde_prefix(self):
|
||||
env = make_env(
|
||||
'(defcomp ~card (&key) (div (~badge)))',
|
||||
'(defcomp ~badge (&key) (span "★"))',
|
||||
)
|
||||
deps = transitive_deps("card", env)
|
||||
assert deps == {"~badge"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# compute_all_deps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestComputeAllDeps:
|
||||
def test_sets_deps_on_components(self):
|
||||
env = make_env(
|
||||
'(defcomp ~page (&key) (div (~card)))',
|
||||
'(defcomp ~card (&key) (div (~badge)))',
|
||||
'(defcomp ~badge (&key) (span "★"))',
|
||||
)
|
||||
compute_all_deps(env)
|
||||
assert env["~page"].deps == {"~card", "~badge"}
|
||||
assert env["~card"].deps == {"~badge"}
|
||||
assert env["~badge"].deps == set()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# scan_components_from_sx
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestScanComponentsFromSx:
|
||||
def test_basic(self):
|
||||
source = '(~card :title "hi" (~badge :label "new"))'
|
||||
refs = scan_components_from_sx(source)
|
||||
assert refs == {"card", "badge"}
|
||||
|
||||
def test_no_components(self):
|
||||
source = '(div :class "p-4" (p "hello"))'
|
||||
refs = scan_components_from_sx(source)
|
||||
assert refs == set()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# components_needed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestComponentsNeeded:
|
||||
def test_page_with_deps(self):
|
||||
env = make_env(
|
||||
'(defcomp ~page-layout (&key) (div (~nav) (~footer)))',
|
||||
'(defcomp ~nav (&key) (nav "nav"))',
|
||||
'(defcomp ~footer (&key) (footer "footer"))',
|
||||
'(defcomp ~unused (&key) (div "not needed"))',
|
||||
)
|
||||
compute_all_deps(env)
|
||||
page_sx = '(~page-layout)'
|
||||
needed = components_needed(page_sx, env)
|
||||
assert "~page-layout" in needed
|
||||
assert "~nav" in needed
|
||||
assert "~footer" in needed
|
||||
assert "~unused" not in needed
|
||||
@@ -167,6 +167,7 @@ class Component:
|
||||
body: Any # unevaluated s-expression body
|
||||
closure: dict[str, Any] = field(default_factory=dict)
|
||||
css_classes: set[str] = field(default_factory=set) # pre-scanned :class values
|
||||
deps: set[str] = field(default_factory=set) # transitive component deps (~names)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<Component ~{self.name}({', '.join(self.params)})>"
|
||||
|
||||
Reference in New Issue
Block a user