wip use_query and use_mutation

Author: rmorshea

docs/features/hooks.md

@@ -4,6 +4,55 @@
 
     Check out the [IDOM Core docs](https://idom-docs.herokuapp.com/docs/reference/hooks-api.html?highlight=hooks) on hooks!
 
+## Use Query and Use Mutation
+
+<!-- TODO: Add description -->
+
+```python
+from example_project.my_app.models import TodoItem
+from idom import component, html
+from django_idom.hooks import use_query, use_mutation
+
+
+def get_items():
+    return TodoItem.objects.all()
+
+def add_item(text: str):
+    TodoItem(text=text).save()
+
+
+@component
+def todo_list():
+    items_query = use_query(get_items)
+    add_item_mutation = use_mutation(add_item, refetch=get_items)
+    item_draft, set_item_draft = use_state("")
+
+    if items_query.loading:
+        items_view = html.h2("Loading...")
+    elif items_query.error:
+        items_view = html.h2(f"Error when loading: {items.error}")
+    else:
+        items_view = html.ul(html.li(item, key=item) for item in items_query.data)
+
+    if add_item_mutation.loading:
+        add_item_status = html.h2("Adding...")
+    elif add_item_mutation.error:
+        add_item_status = html.h2(f"Error when adding: {add_item_mutation.error}")
+    else:
+        add_item_status = ""
+
+    def handle_add_item(event):
+        set_item_draft("")
+        add_item_mutation.execute(text=item_draft)
+
+    return html.div(
+        html.label("Add an item:")
+        html.input({"type": "text", "onClick": handle_add_item})
+        add_item_status,
+        items_view,
+    )
+```
+
 ## Use Websocket
 
 You can fetch the Django Channels websocket at any time by using `use_websocket`.
@@ -18,8 +67,6 @@ def MyComponent():
     return html.div(my_websocket)
 ```
 
-
-
 ## Use Scope
 
 This is a shortcut that returns the Websocket's `scope`.
@@ -34,7 +81,6 @@ def MyComponent():
     return html.div(my_scope)
 ```
 
-
 ## Use Location
 
 ??? info "This hook's behavior will be changed in a future update"

requirements/pkg-deps.txt

@@ -1,3 +1,4 @@
 channels >=3.0.0
 idom >=0.39.0, <0.40.0
 aiofile >=3.0
+typing_extensions

src/django_idom/hooks.py

@@ -1,8 +1,28 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
-from typing import Awaitable, Callable, Dict, Optional, Type, Union
+from threading import Thread
+from types import FunctionType
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    DefaultDict,
+    Optional,
+    Sequence,
+    Type,
+    Union,
+    TypeVar,
+    Generic,
+    NamedTuple,
+)
+
+from typing_extensions import ParamSpec
+from idom import use_callback
 
 from idom.backend.types import Location
-from idom.core.hooks import Context, create_context, use_context
+from idom.core.hooks import Context, create_context, use_context, use_state, use_effect
+from django_idom.utils import UNDEFINED
 
 
 @dataclass
@@ -26,7 +46,7 @@ def use_location() -> Location:
     return Location(scope["path"], f"?{search}" if search else "")
 
 
-def use_scope() -> Dict:
+def use_scope() -> dict[str, Any]:
     """Get the current ASGI scope dictionary"""
     return use_websocket().scope
 
@@ -37,3 +57,116 @@ def use_websocket() -> IdomWebsocket:
     if websocket is None:
         raise RuntimeError("No websocket. Are you running with a Django server?")
     return websocket
+
+
+_REFETCH_CALLBACKS: DefaultDict[FunctionType, set[Callable[[], None]]] = DefaultDict(
+    set
+)
+
+
+_Data = TypeVar("_Data")
+_Params = ParamSpec("_Params")
+
+
+def use_query(
+    query: Callable[_Params, _Data],
+    *args: _Params.args,
+    **kwargs: _Params.kwargs,
+) -> Query[_Data]:
+    given_query = query
+    query, _ = use_state(given_query)
+    if given_query is not query:
+        raise ValueError(f"Query function changed from {query} to {given_query}.")
+
+    data, set_data = use_state(UNDEFINED)
+    loading, set_loading = use_state(True)
+    error, set_error = use_state(None)
+
+    @use_callback
+    def refetch() -> None:
+        set_data(UNDEFINED)
+        set_loading(True)
+        set_error(None)
+
+    @use_effect(dependencies=[])
+    def add_refetch_callback():
+        # By tracking callbacks globally, any usage of the query function will be re-run
+        # if the user has told a mutation to refetch it.
+        _REFETCH_CALLBACKS[query].add(refetch)
+        return lambda: _REFETCH_CALLBACKS[query].remove(refetch)
+
+    @use_effect(dependencies=None)
+    def execute_query():
+        if data is not UNDEFINED:
+            return
+
+        def thread_target():
+            try:
+                returned = query(*args, **kwargs)
+            except Exception as e:
+                set_data(UNDEFINED)
+                set_loading(False)
+                set_error(e)
+            else:
+                set_data(returned)
+                set_loading(False)
+                set_error(None)
+
+        # We need to run this in a thread so we don't prevent rendering when loading.
+        # I'm also hoping that Django is ok with this since this thread won't have an
+        # active event loop.
+        Thread(target=thread_target, daemon=True).start()
+
+    return Query(data, loading, error, refetch)
+
+
+class Query(NamedTuple, Generic[_Data]):
+    data: _Data
+    loading: bool
+    error: Exception | None
+    refetch: Callable[[], None]
+
+
+def use_mutation(
+    mutate: Callable[_Params, None],
+    refetch: Callable[..., Any] | Sequence[Callable[..., Any]],
+) -> Mutation[_Params]:
+    loading, set_loading = use_state(True)
+    error, set_error = use_state(None)
+
+    @use_callback
+    def call(*args: _Params.args, **kwargs: _Params.kwargs) -> None:
+        set_loading(True)
+
+        def thread_target():
+            try:
+                mutate(*args, **kwargs)
+            except Exception as e:
+                set_loading(False)
+                set_error(e)
+            else:
+                set_loading(False)
+                set_error(None)
+                for query in (refetch,) if isinstance(refetch, Query) else refetch:
+                    refetch_callback = _REFETCH_CALLBACKS.get(query)
+                    if refetch_callback is not None:
+                        refetch_callback()
+
+        # We need to run this in a thread so we don't prevent rendering when loading.
+        # I'm also hoping that Django is ok with this since this thread won't have an
+        # active event loop.
+        Thread(target=thread_target, daemon=True).start()
+
+    @use_callback
+    def reset() -> None:
+        set_loading(False)
+        set_error(None)
+
+    return Query(call, loading, error, reset)
+
+
+class Mutation(NamedTuple, Generic[_Params]):
+    execute: Callable[_Params, None]
+    loading: bool
+    error: Exception | None
+    reset: Callable[[], None]

src/django_idom/utils.py

@@ -133,3 +133,12 @@ def _register_components(self, components: Set) -> None:
                     "\033[0m",
                     component,
                 )
+
+
+class _Undefined:
+    def __repr__(self):
+        return "UNDEFINED"
+
+
+UNDEFINED = _Undefined()
+"""Sentinel for undefined values"""