")})"""
+_GENERIC_KWARG_PATTERN = r"""(\s*.*?)"""
+COMMENT_REGEX = re.compile(r"")
+COMPONENT_REGEX = re.compile(
+ r"{%\s*"
+ + _TAG_PATTERN
+ + r"\s*"
+ + _PATH_PATTERN
+ + rf"({_OFFLINE_KWARG_PATTERN}|{_GENERIC_KWARG_PATTERN})*?"
+ + r"\s*%}"
+)
+JINJA_COMPONENT_REGEX = re.compile(
+ r"\{\{\s*"
+ + _TAG_PATTERN
+ + r"\s*\("
+ + r"\s*"
+ + _PATH_PATTERN
+ + rf"({_OFFLINE_KWARG_PATTERN}|{_GENERIC_KWARG_PATTERN})*?"
+ + r"\s*\)\s*\}\}"
+)
+FILE_ASYNC_ITERATOR_THREAD = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ReactPy-Django-FileAsyncIterator")
+SYNC_LAYOUT_THREAD = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ReactPy-Django-SyncLayout")
+
+
+async def render_view(
+ view: Callable | View,
+ request: HttpRequest,
+ args: Sequence,
+ kwargs: dict,
+) -> HttpResponse:
+ """Ingests a Django view (class or function) and returns an HTTP response object."""
+ # Convert class-based view to function-based view
+ if getattr(view, "as_view", None):
+ view = view.as_view() # type: ignore
+
+ # Sync/Async function view
+ response = await ensure_async(view)(request, *args, **kwargs) # type: ignore
+
+ # TemplateView needs an extra render step
+ if getattr(response, "render", None):
+ response = await ensure_async(response.render)()
+
+ return response
+
+
+def register_component(component: ComponentConstructor | str):
+ """Adds a component to the list of known registered components.
+
+ Args:
+ component: The component to register. Can be a component function or dotted path to a component.
+
+ """
+ from reactpy_django.config import (
+ REACTPY_FAILED_COMPONENTS,
+ REACTPY_REGISTERED_COMPONENTS,
+ )
+
+ dotted_path = component if isinstance(component, str) else generate_obj_name(component)
+ try:
+ REACTPY_REGISTERED_COMPONENTS[dotted_path] = import_dotted_path(dotted_path)
+ except AttributeError as e:
+ REACTPY_FAILED_COMPONENTS.add(dotted_path)
+ msg = f"Error while fetching '{dotted_path}'. {(str(e).capitalize())}."
+ raise ComponentDoesNotExistError(msg) from e
+
+
+def register_iframe(view: Callable | View | str):
+ """Registers a view to be used as an iframe component.
+
+ Args:
+ view: The view to register. Can be a function or class based view, or a dotted path to a view.
+ """
+ from reactpy_django.config import REACTPY_REGISTERED_IFRAME_VIEWS
+
+ if hasattr(view, "view_class"):
+ view = view.view_class # type: ignore
+ dotted_path = view if isinstance(view, str) else generate_obj_name(view)
+ try:
+ REACTPY_REGISTERED_IFRAME_VIEWS[dotted_path] = import_dotted_path(dotted_path)
+ except AttributeError as e:
+ msg = f"Error while fetching '{dotted_path}'. {(str(e).capitalize())}."
+ raise ViewDoesNotExistError(msg) from e
+
+
+def import_dotted_path(dotted_path: str) -> Callable:
+ """Imports a dotted path and returns the callable."""
+ module_name, component_name = dotted_path.rsplit(".", 1)
+
+ try:
+ module = import_module(module_name)
+ except ImportError as error:
+ msg = f"Failed to import {module_name!r} while loading {component_name!r}"
+ raise RuntimeError(msg) from error
+
+ return getattr(module, component_name)
+
+
+class RootComponentFinder:
+ """Searches Django templates to find and register all root components.
+ This should only be `run` once on startup to maintain synchronization during mulitprocessing.
+ """
+
+ def run(self):
+ """Registers all ReactPy components found within Django templates."""
+ # Get all template folder paths
+ paths = self.get_paths()
+ # Get all HTML template files
+ templates = self.get_templates(paths)
+ # Get all components
+ components = self.get_components(templates)
+ # Register all components
+ self.register_components(components)
+
+ def get_loaders(self):
+ """Obtains currently configured template loaders."""
+ template_source_loaders = []
+ for e in engines.all():
+ if hasattr(e, "engine"):
+ template_source_loaders.extend(e.engine.get_template_loaders(e.engine.loaders)) # type: ignore
+ loaders = []
+ for loader in template_source_loaders:
+ if hasattr(loader, "loaders"):
+ loaders.extend(loader.loaders)
+ else:
+ loaders.append(loader)
+ return loaders
+
+ def get_paths(self) -> set[str]:
+ """Obtains a set of all template directories."""
+ paths: set[str] = set()
+ for loader in self.get_loaders():
+ with contextlib.suppress(ImportError, AttributeError, TypeError):
+ module = import_module(loader.__module__)
+ get_template_sources = getattr(module, "get_template_sources", None)
+ if get_template_sources is None:
+ get_template_sources = loader.get_template_sources
+ paths.update(smart_str(origin) for origin in get_template_sources(""))
+ return paths
+
+ def get_templates(self, paths: set[str]) -> set[str]:
+ """Obtains a set of all HTML template paths."""
+ extensions = [".html"]
+ templates: set[str] = set()
+ for path in paths:
+ for root, _, files in os.walk(path, followlinks=False):
+ templates.update(
+ os.path.join(root, name)
+ for name in files
+ if not name.startswith(".") and any(fnmatch(name, f"*{glob}") for glob in extensions)
+ )
+
+ return templates
+
+ def get_components(self, templates: set[str]) -> set[str]:
+ """Obtains a set of all ReactPy components by parsing HTML templates."""
+ components: set[str] = set()
+ for template in templates:
+ with contextlib.suppress(Exception), open(template, encoding="utf-8") as template_file:
+ clean_template = COMMENT_REGEX.sub("", template_file.read())
+ regex_iterable = list(COMPONENT_REGEX.finditer(clean_template))
+ regex_iterable.extend(JINJA_COMPONENT_REGEX.finditer(clean_template))
+ new_components: list[str] = []
+ for match in regex_iterable:
+ new_components.append(match.group("path").replace('"', "").replace("'", ""))
+ offline_path = match.group("offline_path")
+ if offline_path:
+ new_components.append(offline_path.replace('"', "").replace("'", ""))
+ components.update(new_components)
+ if not components:
+ _logger.warning(
+ "\033[93m"
+ "ReactPy did not find any components! "
+ "You are either not using any ReactPy components, "
+ "using the template tag incorrectly, "
+ "or your HTML templates are not registered with Django."
+ "\033[0m"
+ )
+ return components
+
+ def register_components(self, components: set[str]) -> None:
+ """Registers all ReactPy components in an iterable."""
+ if components:
+ _logger.debug("Auto-detected ReactPy root components:")
+ for component in components:
+ try:
+ _logger.debug("\t+ %s", component)
+ register_component(component)
+ except Exception:
+ _logger.exception(
+ "\033[91m"
+ "ReactPy failed to register component '%s'!\n"
+ "This component path may not be valid, "
+ "or an exception may have occurred while importing.\n"
+ "See the traceback below for more information."
+ "\033[0m",
+ component,
+ )
+
+
+def generate_obj_name(obj: Any) -> str:
+ """Makes a best effort to create a name for an object.
+ Useful for JSON serialization of Python objects."""
+
+ # First attempt: Create a dotted path by inspecting dunder methods
+ if hasattr(obj, "__module__"):
+ if hasattr(obj, "__name__"):
+ return f"{obj.__module__}.{obj.__name__}"
+ if hasattr(obj, "__class__") and hasattr(obj.__class__, "__name__"):
+ return f"{obj.__module__}.{obj.__class__.__name__}"
+
+ # Second attempt: String representation
+ with contextlib.suppress(Exception):
+ return str(obj)
+
+ # Fallback: Empty string
+ return ""
+
+
+def django_query_postprocessor(
+ data: QuerySet | Model, many_to_many: bool = True, many_to_one: bool = True
+) -> QuerySet | Model:
+ """Recursively fetch all fields within a `Model` or `QuerySet` to ensure they are not performed lazily.
+
+ Behavior can be modified through `postprocessor_kwargs` within your `use_query` hook.
+
+ Args:
+ data: The `Model` or `QuerySet` to recursively fetch fields from.
+
+ Keyword Args:
+ many_to_many: Whether or not to recursively fetch `ManyToManyField` relationships.
+ many_to_one: Whether or not to recursively fetch `ForeignKey` relationships.
+
+ Returns:
+ The `Model` or `QuerySet` with all fields fetched.
+ """
+
+ # `QuerySet`, which is an iterable containing `Model`/`QuerySet` objects.
+ if isinstance(data, QuerySet):
+ for model in data:
+ django_query_postprocessor(
+ model,
+ many_to_many=many_to_many,
+ many_to_one=many_to_one,
+ )
+
+ # `Model` instances
+ elif isinstance(data, Model):
+ prefetch_fields: list[str] = []
+ for field in data._meta.get_fields():
+ # Force the query to execute
+ getattr(data, field.name, None)
+
+ if many_to_one and type(field) is ManyToOneRel:
+ prefetch_fields.append(field.related_name or f"{field.name}_set")
+
+ elif many_to_many and isinstance(field, ManyToManyField):
+ prefetch_fields.append(field.name)
+
+ if prefetch_fields:
+ prefetch_related_objects([data], *prefetch_fields)
+ for field_str in prefetch_fields:
+ django_query_postprocessor(
+ getattr(data, field_str).get_queryset(),
+ many_to_many=many_to_many,
+ many_to_one=many_to_one,
+ )
+
+ # Unrecognized type
+ else:
+ msg = (
+ f"Django query postprocessor expected a Model or QuerySet, got {data!r}.\n"
+ "One of the following may have occurred:\n"
+ " - You are using a non-Django ORM.\n"
+ " - You are attempting to use `use_query` to fetch non-ORM data.\n\n"
+ "If these situations apply, you may want to disable the postprocessor."
+ )
+ raise TypeError(msg)
+
+ return data
+
+
+def validate_component_args(func, *args, **kwargs):
+ """
+ Validate whether a set of args/kwargs would work on the given component.
+
+ Raises `ComponentParamError` if the args/kwargs are invalid.
+ """
+ signature = inspect.signature(func)
+
+ try:
+ signature.bind(*args, **kwargs)
+ except TypeError as e:
+ name = generate_obj_name(func)
+ msg = f"Invalid args for '{name}'. {str(e).capitalize()}."
+ raise ComponentParamError(msg) from e
+
+
+def create_cache_key(*args):
+ """Creates a cache key string that starts with `reactpy_django` contains
+ all *args separated by `:`."""
+
+ if not args:
+ msg = "At least one argument is required to create a cache key."
+ raise ValueError(msg)
+
+ return f"reactpy_django:{':'.join(str(arg) for arg in args)}"
+
+
+class SyncLayout(Layout):
+ """Sync adapter for ReactPy's `Layout`. Allows it to be used in Django template tags.
+ This can be removed when Django supports async template tags.
+ """
+
+ def __enter__(self):
+ self.loop = asyncio.new_event_loop()
+ SYNC_LAYOUT_THREAD.submit(self.loop.run_until_complete, self.__aenter__()).result()
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ SYNC_LAYOUT_THREAD.submit(self.loop.run_until_complete, self.__aexit__(exc_type, exc_val, exc_tb)).result()
+ self.loop.close()
+
+ def sync_render(self):
+ return SYNC_LAYOUT_THREAD.submit(self.loop.run_until_complete, self.render()).result()
+
+
+def get_pk(model):
+ """Returns the value of the primary key for a Django model."""
+ return getattr(model, model._meta.pk.name)
+
+
+def str_to_bool(val: str) -> bool:
+ """Convert a string representation of truth to true (1) or false (0).
+
+ True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
+ are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if
+ 'val' is anything else.
+ """
+ val = val.lower()
+ if val in {"y", "yes", "t", "true", "on", "1"}:
+ return True
+ if val in {"n", "no", "f", "false", "off", "0"}:
+ return False
+ msg = f"invalid truth value {val}"
+ raise ValueError(msg)
+
+
+def prerender_component(
+ user_component: ComponentConstructor,
+ args: Sequence,
+ kwargs: Mapping,
+ uuid: str | UUID,
+ request: HttpRequest,
+) -> str:
+ """Prerenders a ReactPy component and returns the HTML string."""
+ search = request.GET.urlencode()
+ scope = getattr(request, "scope", {})
+ scope["reactpy"] = {"id": str(uuid)}
+ dir(request.user) # Call `dir` before prerendering to make sure the user object is loaded
+
+ with SyncLayout(
+ ConnectionContext(
+ user_component(*args, **kwargs),
+ value=Connection(
+ scope=scope,
+ location=Location(path=request.path, query_string=f"?{search}" if search else ""),
+ carrier=request,
+ ),
+ )
+ ) as layout:
+ vdom_tree = layout.sync_render()["model"]
+
+ return _reactpy_to_string(vdom_tree) # type: ignore
+
+
+def reactpy_to_string(vdom_or_component: Any, request: HttpRequest | None = None, uuid: str | None = None) -> str:
+ """Converts a VdomDict or component to an HTML string. If a string is provided instead, it will be
+ automatically returned."""
+ if isinstance(vdom_or_component, dict):
+ return _reactpy_to_string(vdom_or_component) # type: ignore
+
+ if hasattr(vdom_or_component, "render"):
+ if not request:
+ request = HttpRequest()
+ request.method = "GET"
+ if not uuid:
+ uuid = uuid4().hex
+ return prerender_component(vdom_or_component, [], {}, uuid, request)
+
+ if isinstance(vdom_or_component, str):
+ return vdom_or_component
+
+ msg = f"Invalid type for vdom_or_component: {type(vdom_or_component)}. Expected a VdomDict, component, or string."
+ raise ValueError(msg)
+
+
+def save_component_params(args, kwargs, uuid) -> None:
+ """Saves the component parameters to the database.
+ This is used within our template tag in order to propogate
+ the parameters between the HTTP and WebSocket stack."""
+ from reactpy_django import models
+ from reactpy_django.types import ComponentParams
+
+ params = ComponentParams(args, kwargs)
+ model = models.ComponentSession(uuid=uuid, params=dill.dumps(params))
+ model.full_clean()
+ model.save()
+
+
+def validate_host(host: str) -> None:
+ """Validates the host string to ensure it does not contain a protocol."""
+ if "://" in host:
+ protocol = host.split("://", maxsplit=1)[0]
+ msg = f"Invalid host provided to component. Contains a protocol '{protocol}://'."
+ _logger.error(msg)
+ raise InvalidHostError(msg)
+
+
+class FileAsyncIterator:
+ """Async iterator that yields chunks of data from the provided async file."""
+
+ def __init__(self, file_path: str):
+ self.file_path = file_path
+
+ async def __aiter__(self):
+ file_handle = None
+ try:
+ file_handle = FILE_ASYNC_ITERATOR_THREAD.submit(open, self.file_path, "rb").result()
+ while True:
+ chunk = FILE_ASYNC_ITERATOR_THREAD.submit(file_handle.read, 8192).result()
+ if not chunk:
+ break
+ yield chunk
+ finally:
+ if file_handle:
+ file_handle.close()
+
+
+def ensure_async(
+ func: Callable[FuncParams, Inferred], *, thread_sensitive: bool = True
+) -> Callable[FuncParams, Awaitable[Inferred]]:
+ """Ensure the provided function is always an async coroutine. If the provided function is
+ not async, it will be adapted."""
+
+ @wraps(func)
+ def wrapper(*args, **kwargs):
+ return (
+ func(*args, **kwargs)
+ if iscoroutinefunction(func)
+ else database_sync_to_async(func, thread_sensitive=thread_sensitive)(*args, **kwargs)
+ )
+
+ return wrapper
+
+
+def cached_static_file(static_path: str) -> str:
+ from reactpy_django.config import REACTPY_CACHE
+
+ # Try to find the file within Django's static files
+ abs_path = find(static_path)
+ if not abs_path:
+ msg = f"Could not find static file {static_path} within Django's static files."
+ raise FileNotFoundError(msg)
+ if isinstance(abs_path, (list, tuple)):
+ abs_path = abs_path[0]
+
+ # Fetch the file from cache, if available
+ last_modified_time = os.stat(abs_path).st_mtime
+ cache_key = f"reactpy_django:static_contents:{static_path}"
+ file_contents: str | None = caches[REACTPY_CACHE].get(cache_key, version=int(last_modified_time))
+ if file_contents is None:
+ with open(abs_path, encoding="utf-8") as static_file:
+ file_contents = static_file.read()
+ caches[REACTPY_CACHE].delete(cache_key)
+ caches[REACTPY_CACHE].set(cache_key, file_contents, timeout=None, version=int(last_modified_time))
+
+ return file_contents
+
+
+def del_html_head_body_transform(vdom: VdomDict) -> VdomDict:
+ """Transform intended for use with `string_to_reactpy `.
+
+ Removes ``, ``, and `` while preserving their children.
+
+ Parameters:
+ vdom:
+ The VDOM dictionary to transform.
+ """
+ if vdom["tagName"] in {"html", "body", "head"}:
+ return VdomDict(tagName="", children=vdom.setdefault("children", []))
+ return vdom
+
+
+def fetch_cached_python_file(file_path: str, minify: bool = True) -> str:
+ from reactpy.executors.pyscript.utils import minify_python
+
+ from reactpy_django.config import REACTPY_CACHE
+
+ # Try to get user code from cache
+ cache_key = create_cache_key("pyscript", file_path)
+ last_modified_time = os.stat(file_path).st_mtime
+ file_contents: str = caches[REACTPY_CACHE].get(cache_key, version=int(last_modified_time))
+ if file_contents:
+ return file_contents
+
+ file_contents = Path(file_path).read_text(encoding="utf-8").strip()
+ if minify:
+ file_contents = minify_python(file_contents)
+ caches[REACTPY_CACHE].set(cache_key, file_contents, version=int(last_modified_time))
+ return file_contents
diff --git a/tests/test_app/__init__.py b/tests/test_app/__init__.py
index df44ad6c..cccc67a5 100644
--- a/tests/test_app/__init__.py
+++ b/tests/test_app/__init__.py
@@ -5,15 +5,20 @@
# Make sure the JS is always re-built before running the tests
js_dir = Path(__file__).parent.parent.parent / "src" / "js"
static_dir = Path(__file__).parent.parent.parent / "src" / "reactpy_django" / "static" / "reactpy_django"
-assert subprocess.run(["bun", "install"], cwd=str(js_dir), check=True).returncode == 0
-assert (
+
+# Check if bun is available; if so, rebuild the JS. If not, assume artifacts exist.
+_bun_available = shutil.which("bun") is not None
+if _bun_available:
+ subprocess.run(["bun", "install"], cwd=str(js_dir), check=True)
subprocess.run(
["bun", "build", "./src/index.ts", f"--outdir={static_dir}", "--sourcemap=linked"],
cwd=str(js_dir),
check=True,
- ).returncode
- == 0
-)
+ )
+# Verify that JS artifacts already exist so we don't silently skip a needed build
+elif not (static_dir / "index.js").exists():
+ msg = f"bun is not installed and JS artifacts are missing. Run 'bun install && bun build' in {js_dir} first."
+ raise RuntimeError(msg)
# Make sure the test environment is always using the latest JS
diff --git a/tests/test_app/jinja2_templates/base.html b/tests/test_app/jinja2_templates/base.html
new file mode 100644
index 00000000..90aa4068
--- /dev/null
+++ b/tests/test_app/jinja2_templates/base.html
@@ -0,0 +1,97 @@
+
+
+
+
+ ReactPy
+
+
+
+
+ ReactPy Test Page (Jinja2)
+ ReactPy
+
+
+
+ ReactPy Errors Test Page (Jinja2)
+ {{ component("test_app.components.does_not_exist") }}
+ {{ component("test_app.components.hello_world", invalid_param="random_value") }}
+ {{ component("test_app.components.hello_world", host="https://example.com/") }}
+
+ {{ component("test_app.components.broken_postprocessor_query") }}
+
+
+ {{ component("test_app.components.view_to_iframe_not_registered") }}
+
+
+ {{ component("test_app.components.incorrect_user_passes_test_decorator") }}
+ ReactPy
+
+
+
+ ReactPy Test Page
+ Custom Host ({{ new_host }}):
+ {{ component("test_app.components.custom_host", host=new_host, number=0) }}
+ ReactPy
+
+
+
+ ReactPy Test Page
+ Round-Robin Host:
+ {{ component("test_app.components.custom_host", number=count_val) }}
+ ReactPy
+
+
+
+ ReactPy Offline Test Page
+ ReactPy
+
+
+
+ ReactPy Prerender Test Page
+
+ {{ component("test_app.prerender.components.prerender_string", class="prerender-string", prerender="true") }}
+
+
+ {{ component("test_app.prerender.components.prerender_vdom", class="prerender-vdom", prerender="true") }}
+
+
+ {{ component("test_app.prerender.components.prerender_component", class="prerender-component", prerender="true") }}
+
+