"""Discover co-located CSS/JS files and page and component module asset lists.
This module owns the filesystem side of the static pipeline. It walks
layout chains, reads `styles` and `scripts` module lists, and pushes
results onto a collector via the active backend.
The path-to-logical-name conversion lives on the `PathResolver` so both
discovery and the staticfiles finder share the exact same mapping. The
`StemRegistry` controls which filenames are auto-picked-up per role. It
lets users teach the framework about new stems like `page.css` or
`panel.js` without patching the core.
The `BackendProvider` protocol inverts the dependency direction. The
discovery layer does not import the static manager directly. Any object
exposing `default_backend` and `page_roots` satisfies the protocol,
which makes unit-testing without a full manager trivial.
The provider contract requires that `page_roots` returns already
resolved absolute paths. Both the static manager and the co-located
finder satisfy this contract, which lets the discovery layer skip a
round of resolution on every logical-name lookup.
"""
from __future__ import annotations
import logging
import os
from collections import OrderedDict
from typing import TYPE_CHECKING, Protocol, runtime_checkable
from next.pages import loaders as pages_loaders
from .assets import StaticAsset, default_kinds
from .collector import default_placeholders
from .signals import asset_registered
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from next.components import ComponentInfo
from .backends import StaticBackend
from .collector import StaticCollector
logger = logging.getLogger(__name__)
_MODULE_LIST_CACHE_MAX_SIZE = 2048
_LAYOUT_DIR_CACHE_MAX_SIZE = 2048
def _url_suffix(url: str) -> str:
"""Return the lowercase dot-suffix of a URL or empty string when absent."""
last_segment = url.rsplit("?", 1)[0].rsplit("#", 1)[0].rsplit("/", 1)[-1]
dot = last_segment.rfind(".")
if dot < 0:
return ""
return last_segment[dot:].lower()
def _rel_path_str(child: Path, root: Path) -> str | None:
"""Return the forward-slashed path of `child` relative to `root`.
Both operands must already be resolved absolute paths. Returns an
empty string when `child == root`, and `None` when `child` is not
nested under `root`. Skips the per-segment work of
`Path.relative_to().parts`.
"""
child_str = os.fspath(child)
root_str = os.fspath(root)
if child_str == root_str:
return ""
prefix = root_str + os.sep
if not child_str.startswith(prefix): # pragma: no cover
return None
rel = child_str[len(prefix) :]
if os.sep != "/": # pragma: no cover
rel = rel.replace(os.sep, "/")
return rel
[docs]
@runtime_checkable
class BackendProvider(Protocol):
"""Contract consumed by the asset discovery layer.
The static manager is the canonical implementation. Tests can pass
any object exposing `default_backend` and `page_roots` without
instantiating the full manager. Implementations must return
resolved absolute paths from `page_roots`.
"""
@property
def default_backend(self) -> StaticBackend:
"""Return the primary backend used for file registration."""
raise NotImplementedError
[docs]
def page_roots(self) -> tuple[Path, ...]:
"""Return the configured page-tree roots as resolved absolute paths."""
raise NotImplementedError
[docs]
class StemRegistry:
"""Map discovery role to registered filename stems.
Default roles and their stems are as follows. The `template` role
maps to `["template"]` and matches `template.css` or `template.js`.
The `layout` role maps to `["layout"]` and matches `layout.css` or
`layout.js`. The `component` role maps to `["component"]` and
matches `component.css`.
Users may register extra stems during `AppConfig.ready` to teach
discovery about new filenames.
Example::
default_stems.register("template", "page")
The example above teaches discovery to also pick up `page.css` or
`page.js` alongside `template.css` or `template.js`.
"""
DEFAULT_ROLES: tuple[str, ...] = ("template", "layout", "component")
[docs]
def __init__(self) -> None:
"""Seed the registry with the built-in template, layout, and component roles."""
self._roles: dict[str, list[str]] = {
"template": ["template"],
"layout": ["layout"],
"component": ["component"],
}
[docs]
def register(self, role: str, stem: str) -> None:
"""Add a stem under the given role, creating the role when missing."""
stems = self._roles.setdefault(role, [])
if stem not in stems:
stems.append(stem)
[docs]
def stems(self, role: str) -> tuple[str, ...]:
"""Return registered stems for the role in registration order."""
return tuple(self._roles.get(role, ()))
[docs]
def roles(self) -> tuple[str, ...]:
"""Return all registered roles in registration order."""
return tuple(self._roles)
default_stems: StemRegistry = StemRegistry()
[docs]
class PathResolver:
"""Resolve page root and logical names for page, layout, and component paths.
The resolver is shared between the asset discovery layer and the
staticfiles finder so both layers produce identical logical names
for the same on-disk location. The resolver assumes that the
provider callable returns already resolved absolute page roots.
"""
[docs]
def __init__(
self,
page_roots_provider: Callable[[], tuple[Path, ...]],
) -> None:
"""Store the page-roots provider callable consulted on every lookup."""
self._provider = page_roots_provider
self._find_page_root_cache: dict[Path, Path | None] = {}
[docs]
def page_roots(self) -> tuple[Path, ...]:
"""Return the current tuple of page tree roots from the provider."""
return self._provider()
[docs]
def find_page_root(self, path: Path) -> Path | None:
"""Return the page tree root that contains the path, or None."""
cached = self._find_page_root_cache.get(path)
if cached is not None or path in self._find_page_root_cache:
return cached
resolved_parent = path.parent.resolve()
for root in self.page_roots():
if resolved_parent.is_relative_to(root):
self._find_page_root_cache[path] = root
return root
self._find_page_root_cache[path] = None
return None
[docs]
def logical_name_for_template(
self,
template_dir: Path,
page_root: Path | None,
) -> str:
"""Return the logical URL name for a page template directory.
The caller is expected to pass a resolved `template_dir` and a
resolved `page_root` from `find_page_root`.
"""
if page_root is None:
return self._fallback(template_dir)
rel = _rel_path_str(template_dir, page_root)
if rel is None: # pragma: no cover
return self._fallback(template_dir)
return rel or "index"
[docs]
def logical_name_for_layout(
self,
layout_dir: Path,
page_root: Path | None,
) -> str:
"""Return the logical URL name for a layout directory.
The caller is expected to pass a resolved `layout_dir` and a
resolved `page_root` from `find_page_root`.
"""
if page_root is None:
return f"{self._fallback(layout_dir)}/layout"
rel = _rel_path_str(layout_dir, page_root)
if rel is None: # pragma: no cover
return f"{self._fallback(layout_dir)}/layout"
return f"{rel}/layout" if rel else "layout"
@staticmethod
def _fallback(directory: Path) -> str:
return directory.name or "index"
[docs]
class AssetDiscovery:
"""Detect co-located asset files and module-level asset list variables.
The `provider` argument supplies the active backend and the page
tree roots. The optional `resolver` argument is a path resolver.
The default resolver is backed by the provider. The optional
`stems` argument is a stem registry. The default is the
process-wide `default_stems`.
"""
[docs]
def __init__(
self,
provider: BackendProvider,
*,
resolver: PathResolver | None = None,
stems: StemRegistry | None = None,
) -> None:
"""Bind the provider and wire optional resolver and stems."""
self._provider = provider
self._resolver = resolver or PathResolver(provider.page_roots)
self._stems = stems or default_stems
self._module_list_cache: OrderedDict[Path, dict[str, list[str]]] = OrderedDict()
self._layout_dir_cache: OrderedDict[Path, list[Path]] = OrderedDict()
[docs]
def discover_page_assets(
self,
file_path: Path,
collector: StaticCollector,
) -> None:
"""Collect layout, template, and module-level assets for a page file.
Assets are added from the outermost layout inward, then from
the template directory, then from `styles` and `scripts`
module lists declared in `page.py`.
"""
resolved = file_path.resolve()
page_root = self._resolver.find_page_root(resolved)
for layout_dir in self._find_layout_directories(resolved, page_root):
self._collect_role_directory(
layout_dir,
logical_name=self._resolver.logical_name_for_layout(
layout_dir, page_root
),
role="layout",
collector=collector,
)
self._collect_role_directory(
resolved.parent,
logical_name=self._resolver.logical_name_for_template(
resolved.parent, page_root
),
role="template",
collector=collector,
)
if resolved.exists():
self._collect_module_lists(resolved, collector)
[docs]
def discover_component_assets(
self,
info: ComponentInfo,
collector: StaticCollector,
) -> None:
"""Collect co-located CSS, JS, and module asset lists for a component."""
component_dir = self._component_directory(info)
if component_dir is None:
return
logical_name = f"components/{info.name}"
self._collect_role_directory(
component_dir,
logical_name=logical_name,
role="component",
collector=collector,
)
module_path = info.module_path
if module_path is not None and module_path.exists():
self._collect_module_lists(module_path, collector)
def _collect_role_directory(
self,
directory: Path,
*,
logical_name: str,
role: str,
collector: StaticCollector,
) -> None:
"""Register `{stem}{ext}` files for every registered kind.
The set of extensions probed comes from `KindRegistry.kinds()`,
so registering a new kind during `AppConfig.ready` is enough to
teach discovery about additional file types.
"""
for stem in self._stems.stems(role):
for kind in default_kinds.kinds():
suffix = default_kinds.extension(kind)
candidate = directory / f"{stem}{suffix}"
if candidate.exists():
self._register_file(candidate, logical_name, kind, collector)
def _collect_module_lists(
self,
module_path: Path,
collector: StaticCollector,
) -> None:
"""Read URL list variables matching every registered placeholder slot.
The discovery layer iterates over registered slots in
`default_placeholders` and reads the variable named after each
slot. Each URL gets a `kind` derived from its file extension via
`KindRegistry.kind_for_extension`. URLs whose suffix is not in
the registry are dropped with a debug log.
The caller in `discover_page_assets` passes a resolved module
path. The component entry point still calls with the raw path,
so the key is normalised here as a safety net.
"""
cache_key = module_path if module_path.is_absolute() else module_path.resolve()
if cache_key in self._module_list_cache:
self._module_list_cache.move_to_end(cache_key)
cached = self._module_list_cache[cache_key]
else:
module = pages_loaders._load_python_module(module_path)
if module is None:
self._module_list_cache[cache_key] = {}
if len(self._module_list_cache) > _MODULE_LIST_CACHE_MAX_SIZE:
self._module_list_cache.popitem(last=False)
return
cached = {
slot.name: pages_loaders._read_string_list(module, slot.name)
for slot in default_placeholders
}
self._module_list_cache[cache_key] = cached
if len(self._module_list_cache) > _MODULE_LIST_CACHE_MAX_SIZE:
self._module_list_cache.popitem(last=False)
for slot_name, urls in cached.items():
for url in urls:
self._register_module_url(url, slot_name, collector)
def _register_module_url(
self,
url: str,
slot_name: str,
collector: StaticCollector,
) -> None:
"""Resolve a module-level URL to a kind and add it to the collector.
The kind comes from `KindRegistry.kind_for_extension(suffix)`
where suffix is the lowercase trailing dot-extension of the
URL. URLs whose extension is not registered, or whose resolved
kind belongs to a different slot, are dropped with a debug log.
"""
suffix = _url_suffix(url)
if not suffix:
logger.debug("Module URL %r has no recognised extension", url)
return
kind = default_kinds.kind_for_extension(suffix)
if kind is None:
logger.debug("Module URL %r has unregistered extension %r", url, suffix)
return
if default_kinds.slot(kind) != slot_name:
logger.debug(
"Module URL %r maps to kind %r but list %r expects slot %r",
url,
kind,
slot_name,
slot_name,
)
return
collector.add(StaticAsset(url=url, kind=kind))
def _register_file(
self,
source_path: Path,
logical_name: str,
kind: str,
collector: StaticCollector,
) -> None:
"""Register a file with the backend and add the result to the collector.
Warnings are logged for `OSError` and `ValueError`. All other
exception types propagate so bugs in custom backends surface
loudly.
"""
backend = self._provider.default_backend
try:
url = backend.register_file(source_path, logical_name, kind)
except (OSError, ValueError) as e:
logger.warning(
"Failed to register static asset %s as %r: %s",
source_path,
logical_name,
e,
extra={"source_path": str(source_path), "kind": kind},
)
return
asset = StaticAsset(url=url, kind=kind, source_path=source_path.resolve())
collector.add(asset)
asset_registered.send(
sender=asset,
collector=collector,
backend=backend,
)
def _component_directory(self, info: ComponentInfo) -> Path | None:
"""Return the directory that holds a composite component, or None."""
if info.is_simple:
return None
if info.template_path is not None:
return info.template_path.parent
if info.module_path is not None: # pragma: no cover
return info.module_path.parent
return None # pragma: no cover
def _find_layout_directories(
self,
file_path: Path,
page_root: Path | None,
) -> list[Path]:
"""Walk up from the page directory and return layout dirs outermost first.
The caller is expected to pass a resolved absolute `file_path`.
The resolver contract also guarantees that `page_root` is
resolved, which lets this loop compare paths with `==` without
issuing another filesystem call per iteration.
"""
if file_path in self._layout_dir_cache:
self._layout_dir_cache.move_to_end(file_path)
return self._layout_dir_cache[file_path]
directories: list[Path] = []
current_dir = file_path.parent
while True:
if (current_dir / "layout.djx").exists():
directories.append(current_dir)
if page_root is not None and current_dir == page_root:
break
parent = current_dir.parent
if parent == current_dir:
break
current_dir = parent
result = list(reversed(directories))
self._layout_dir_cache[file_path] = result
if len(self._layout_dir_cache) > _LAYOUT_DIR_CACHE_MAX_SIZE:
self._layout_dir_cache.popitem(last=False)
return result
next.dj