Skip to content

Commit

Permalink
Merge branch 'feature/client_data'
Browse files Browse the repository at this point in the history
  • Loading branch information
Alyxion committed Apr 5, 2024
2 parents 6a24fb6 + fac0356 commit 28c507f
Show file tree
Hide file tree
Showing 4 changed files with 86 additions and 12 deletions.
2 changes: 2 additions & 0 deletions examples/modularization/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import theme

from nicegui import app, ui
from nicegui.single_page import SinglePageRouter


# here we use our custom page decorator directly and just put the content creation into a separate function
Expand All @@ -20,4 +21,5 @@ def index_page() -> None:
# we can also use the APIRouter as described in https://nicegui.io/documentation/page#modularize_with_apirouter
app.include_router(example_c.router)

spr = SinglePageRouter("/").setup_page_routes() # TODO Experimental, for performance comparison
ui.run(title='Modularization Example')
5 changes: 3 additions & 2 deletions nicegui/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,12 @@

from . import background_tasks, binding, core, helpers, json
from .awaitable_response import AwaitableResponse
from .context import get_client
from .dependencies import generate_resources
from .element import Element
from .favicon import get_favicon_url
from .logging import log
from .observables import ObservableDict
from .outbox import Outbox
from .version import __version__

Expand Down Expand Up @@ -79,7 +81,7 @@ def __init__(self, page: page, *, shared: bool = False) -> None:
self._body_html = ''

self.page = page
self.state = {}
self.state = ObservableDict()

self.connect_handlers: List[Union[Callable[..., Any], Awaitable]] = []
self.disconnect_handlers: List[Union[Callable[..., Any], Awaitable]] = []
Expand All @@ -100,7 +102,6 @@ def is_auto_index_client(self) -> bool:
@staticmethod
def current_client() -> Optional[Client]:
"""Returns the current client if obtainable from the current context."""
from .context import get_client
return get_client()

@property
Expand Down
2 changes: 1 addition & 1 deletion nicegui/page_layout.py
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,7 @@ def __init__(self, position: PageStickyPositions = 'bottom-right', x_offset: flo

def _check_current_slot(element: Element) -> None:
parent = context.get_slot().parent
if parent != parent.client.content:
if parent != parent.client.content and parent != parent.client.state.get("__singlePageContent", None):
log.warning(f'Found top level layout element "{element.__class__.__name__}" inside element "{parent.__class__.__name__}". '
'Top level layout elements should not be nested but must be direct children of the page content. '
'This will be raising an exception in NiceGUI 1.5') # DEPRECATED
89 changes: 80 additions & 9 deletions nicegui/single_page.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
import inspect
import urllib.parse
from typing import Callable, Dict, Union, Optional, Tuple

from fastapi import HTTPException
from fastapi.routing import APIRoute

from nicegui import background_tasks, helpers, ui, core, Client, app

SPR_PAGE_BODY = '__pageContent'
SPR_PAGE_BODY = '__singlePageContent'


class SinglePageRouterFrame(ui.element, component='single_page.js'):
Expand Down Expand Up @@ -34,6 +37,65 @@ def __init__(self, path: str, builder: Callable, title: Union[str, None] = None)
self.title = title


class UrlParameterResolver:
"""The UrlParameterResolver is a helper class which is used to resolve the path and query parameters of an URL to
find the matching SinglePageRouterEntry and convert the parameters to the expected types of the builder function"""

def __init__(self, routes: Dict[str, SinglePageRouterEntry], path: str):
"""
:param routes: The routes of the single page router
:param path: The path of the URL
"""
components = path.split("?")
path = components[0].rstrip("/")
self.routes = routes
self.query_string = components[1] if len(components) > 1 else ""
self.query_args = {}
self.path = path
self.path_args = {}
self.parse_query()
self.entry = self.resolve_path()
if self.entry is not None:
self.convert_arguments()

def resolve_path(self) -> Optional[SinglePageRouterEntry]:
"""Splits the path into its components, tries to match it with the routes and extracts the path arguments
into their corresponding variables.
"""
for route, entry in self.routes.items():
route_elements = route.lstrip('/').split("/")
path_elements = self.path.lstrip('/').split("/")
if len(route_elements) != len(path_elements): # can't match
continue
match = True
for i, route_element_path in enumerate(route_elements):
if route_element_path.startswith("{") and route_element_path.endswith("}") and len(
route_element_path) > 2:
self.path_args[route_element_path[1:-1]] = path_elements[i]
elif path_elements[i] != route_element_path:
match = False
break
if match:
return entry
return None

def parse_query(self):
"""Parses the query string of the URL into a dictionary of key-value pairs"""
self.query_args = urllib.parse.parse_qs(self.query_string)

def convert_arguments(self):
"""Converts the path and query arguments to the expected types of the builder function"""
sig = inspect.signature(self.entry.builder)
for name, param in sig.parameters.items():
for params in [self.path_args, self.query_args]:
if name in params:
# Convert parameter to the expected type
try:
params[name] = param.annotation(params[name])
except ValueError as e:
raise ValueError(f"Could not convert parameter {name}: {e}")


class SinglePageRouter:
"""The SinglePageRouter allows the development of a Single Page Application (SPA) which maintains a
persistent connection to the server and only updates the content of the page instead of reloading the whole page.
Expand Down Expand Up @@ -127,7 +189,7 @@ def add_page(self, path: str, builder: Callable, title: Optional[str] = None) ->
:param builder: The builder function
:param title: Optional title of the page
"""
self.routes[path] = SinglePageRouterEntry(path, builder, title)
self.routes[path] = SinglePageRouterEntry(path.rstrip("/"), builder, title)

def add_router_entry(self, entry: SinglePageRouterEntry) -> None:
"""Adds a fully configured SinglePageRouterEntry to the router
Expand All @@ -136,7 +198,7 @@ def add_router_entry(self, entry: SinglePageRouterEntry) -> None:
"""
self.routes[entry.path] = entry

def get_router_entry(self, target: Union[Callable, str]) -> Union[SinglePageRouterEntry, None]:
def get_router_entry(self, target: Union[Callable, str]) -> Tuple[Optional[SinglePageRouterEntry], dict, dict]:
"""Returns the SinglePageRouterEntry for the given target URL or builder function
:param target: The target URL or builder function
Expand All @@ -145,9 +207,14 @@ def get_router_entry(self, target: Union[Callable, str]) -> Union[SinglePageRout
if isinstance(target, Callable):
for path, entry in self.routes.items():
if entry.builder == target:
return entry
return entry, {}, {}
else:
return self.routes.get(target, None)
target = target.rstrip("/")
entry = self.routes.get(target, None)
if entry is None:
parser = UrlParameterResolver(self.routes, target)
return parser.entry, parser.path_args, parser.query_args
return entry, {}, {}

def open(self, target: Union[Callable, str, Tuple[str, bool]]) -> None:
"""Open a new page in the browser by exchanging the content of the root page's slot element
Expand All @@ -158,21 +225,24 @@ def open(self, target: Union[Callable, str, Tuple[str, bool]]) -> None:
target, server_side = target # unpack the list
else:
server_side = True
entry = self.get_router_entry(target)
entry, route_args, query_args = self.get_router_entry(target)
if entry is None:
entry = ui.label(f"Page not found: {target}").classes("text-red-500") # Could be beautified
title = entry.title if entry.title is not None else core.app.config.title
ui.run_javascript(f'document.title = "{title}"')
if server_side:
ui.run_javascript(f'window.history.pushState({{page: "{target}"}}, "", "{target}");')

async def build(content_element) -> None:
async def build(content_element, kwargs) -> None:
with content_element:
result = entry.builder()
result = entry.builder(**kwargs)
if helpers.is_coroutine_function(entry.builder):
await result

content = app.storage.session[SPR_PAGE_BODY]
content.clear()
background_tasks.create(build(content))
combined_dict = {**route_args, **query_args}
background_tasks.create(build(content, combined_dict))

def _find_api_routes(self):
"""Find all API routes already defined via the @page decorator, remove them and redirect them to the
Expand All @@ -186,6 +256,7 @@ def _find_api_routes(self):
title = None
if key in Client.page_configs:
title = Client.page_configs[key].title
route = route.rstrip("/")
self.routes[route] = SinglePageRouterEntry(route, builder=key, title=title)
for route in core.app.routes.copy():
if isinstance(route, APIRoute):
Expand Down

0 comments on commit 28c507f

Please sign in to comment.