### Test State Value Setting and Retrieval

Source: https://context7_llms

Demonstrates setting and getting values from a State object. It verifies that the state can be updated and the new value is correctly retrieved. This is fundamental for state management.

```python
def test_state_set():
    s = State(0)
    s.set(1)
    assert s.get() == 1
```

--------------------------------

### Python Environment Setup for HMR

Source: https://context7_llms

This snippet demonstrates setting up a temporary Python environment using `tempfile.TemporaryDirectory` and `contextlib.chdir`. It manipulates `sys.path` and `sys.meta_path` to include a `ReactiveModuleFinder` and cleans up the environment upon exiting the context manager. This is useful for testing HMR functionality in isolation.

```python
import sys
from collections.abc import Callable
from contextlib import chdir, contextmanager
from tempfile import TemporaryDirectory

from reactivity.hmr.core import ReactiveModuleFinder
from reactivity.hmr.fs import _filters

from .fs import FsUtils
from .io import StringIOWrapper, capture_stdout
from .mock import MockReloader


def compose[T1, T2, **P](first: Callable[P, T1], second: Callable[[T1], T2]) -> Callable[P, T2]:
    """to borrow the params from the first function and the return type from the second one"""
    return lambda *args, **kwargs: second(first(*args, **kwargs))


class Environment(FsUtils):
    def __init__(self, stdout: StringIOWrapper):
        self._stdout = stdout

    @property
    def stdout_delta:
        return self._stdout.delta

    @property
    def hmr:
        def use(reloader: MockReloader):
            """so that using these methods does trigger watchfiles events"""
            self.replace = reloader.replace
            self.write = reloader.write
            return reloader

        return compose(MockReloader, lambda reloader: use(reloader).hmr())

    def __repr__(self):
        return f"Environment(stdout={self._stdout!r})"


@contextmanager
def environment():
    with TemporaryDirectory() as tmpdir, chdir(tmpdir), capture_stdout() as stdout:
        sys.path.append(tmpdir)
        names = {*sys.modules}
        sys.meta_path.insert(0, finder := ReactiveModuleFinder())
        try:
            yield Environment(stdout)
        finally:
            sys.path.remove(tmpdir)
            for name in {*sys.modules} - names:
                del sys.modules[name]
            sys.meta_path.remove(finder)
            _filters.clear()
```

--------------------------------

### Python HMR Context Setup (_common.py)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

This Python module initializes a global HMR context using `new_context` from the `context` module. This `HMR_CONTEXT` is likely used throughout the HMR library for managing state and operations related to hot module reloading.

```python
from ..context import new_context

HMR_CONTEXT = new_context()
```

--------------------------------

### HMR Utilities Setup

Source: https://context7_llms

Imports necessary modules for HMR utilities, including `ast` for parsing, `UserDict` for custom dictionary behavior, `Callable` for type hinting, and `wraps` for decorator utility. These are foundational imports for building more complex HMR functionalities within the `reactivity.hmr` package.

```python
from ast import parse
from collections import UserDict
from collections.abc import Callable
from functools import wraps
```

--------------------------------

### Using Reactivity Under HMR

Source: https://context7_llms

Demonstrates the usage of reactivity primitives like `create_signal` and `create_effect` within a module managed by HMR. This test setup is intended to verify that reactive updates function correctly during HMR.

```python
def test_using_reactivity_under_hmr():
    with environment() as env:

        def simple_test():
            from reactivity import create_effect, create_signal
            from utils import capture_stdout

            get_s, set_s = create_signal(0)

            with capture_stdout() as stdout, create_effect(lambda: print(get_s())):
                # Further assertions would go here to test the effect
                pass

        # The environment setup for this test case is incomplete in the provided snippet.
        
```

--------------------------------

### Run Python Module with HMR

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

Executes a Python module with Hot Module Replacement enabled. It finds the module's entry point, sets up the reactive module, and starts the watching process. Dependencies include `importlib.util.find_spec`, `sys`, `Path`, and HMR core components.

```python
import sys
from pathlib import Path
from importlib.util import find_spec
from types import ModuleType
import builtins

from .core import ReactiveModule, SyncReloader, _loader, patch_meta_path

def run_module(module_name: str, args: list[str]):
    if (cwd := str(Path.cwd())) not in sys.path:
        sys.path.insert(0, cwd)

    patch_meta_path()

    spec = find_spec(module_name)
    if spec is None:
        raise ModuleNotFoundError(f"No module named '{module_name}'")  # noqa: TRY003

    entry: str
    if spec.submodule_search_locations is not None:
        # It's a package, look for __main__.py
        spec = find_spec(f"{module_name}.__main__")
        if spec and spec.origin:
            entry = spec.origin
        else:
            raise ModuleNotFoundError(f"No module named '{module_name}.__main__'; '{module_name}' is a package and cannot be directly executed")  # noqa: TRY003
    elif spec.origin is None:
        raise ModuleNotFoundError(f"Cannot find entry point for module '{module_name}'")  # noqa: TRY003
    else:
        entry = spec.origin

    args[0] = entry  # Replace the first argument with the full path

    _argv = sys.argv[:]
    sys.argv[:] = args
    _main = sys.modules["__main__"]
    try:
        reloader = SyncReloader(entry)
        if spec.loader is not _loader:
            spec.loader = _loader  # make it reactive
        namespace = {"__file__": entry, "__name__": "__main__", "__spec__": spec, "__loader__": _loader, "__package__": spec.parent, "__cached__": None, "__builtins__": builtins}
        sys.modules["__main__"] = reloader.entry_module = ReactiveModule(Path(entry), namespace, "__main__")
        reloader.keep_watching_until_interrupt()
    finally:
        sys.argv[:] = _argv
        sys.modules["__main__"] = _main
```

--------------------------------

### Signal Update vs. Set/Get Tracking in Python

Source: https://context7_llms

Compares the behavior of `s.update()` versus `s.set()` and `s.get()` within `Effect` contexts. It highlights how `update` does not track, while `set`/`get` interactions within an `Effect` trigger re-evaluation, with `Batch.flush` providing deduplication. Relies on `warns` and `Effect` from `reactivity`.

```python
def test_update_vs_set_get_tracking():
    s = Signal(0)

    with warns(RuntimeWarning) as record, Effect(lambda: s.update(lambda x: x + 1)) as e:
        assert record[0].lineno == current_lineno() - 1
        assert s.get() == 1
        assert e not in s.subscribers  # update doesn't track

    # without `.update()`, effects will invalidate themselves, which is unintended mostly
    with Effect(lambda: s.set(s.get() + 1)) as e:
        assert s.get() == 3
        assert e in s.subscribers
        s.set(4)
        assert s.get() == 5  # effect triggered only once because `Batch.flush` has deduplication logic
```

--------------------------------

### API for Synchronous and Asynchronous Reloader in Python

Source: https://context7_llms

This module provides API classes (`SyncReloaderAPI`, `AsyncReloaderAPI`) for integrating HMR into Python applications. It manages the lifecycle of the reloader, including running entry files with hooks, starting and stopping watching, and cleaning up resources. It uses context managers (`__enter__`, `__exit__`, `__aenter__`, `__aexit__`) for convenient setup and teardown.

```python
import sys

from .core import HMR_CONTEXT, AsyncReloader, BaseReloader, SyncReloader
from .hooks import call_post_reload_hooks, call_pre_reload_hooks


class LifecycleMixin(BaseReloader):
    def run_with_hooks(self):
        self._original_main_module = sys.modules["__main__"]
        sys.modules["__main__"] = self.entry_module
        call_pre_reload_hooks()
        self.effect = HMR_CONTEXT.effect(self.run_entry_file)
        call_post_reload_hooks()

    def clean_up(self):
        self.effect.dispose()
        self.entry_module.load.dispose()
        self.entry_module.load.invalidate()
        sys.modules["__main__"] = self._original_main_module


class SyncReloaderAPI(SyncReloader, LifecycleMixin):
    def __enter__(self):
        from threading import Thread

        self.run_with_hooks()
        self.thread = Thread(target=self.start_watching)
        self.thread.start()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(to_thread(self.start_watching))
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()


class AsyncReloaderAPI(AsyncReloader, LifecycleMixin):
    def __enter__(self):
        from asyncio import run
        from threading import Event, Thread

        self.run_with_hooks()

        e = Event()

        async def task():
            e.set()
            await self.start_watching()

        self.thread = Thread(target=lambda: run(task()))
        self.thread.start()
        e.wait()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(self.start_watching())
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()
```

--------------------------------

### Python Package Initialization (reactivity/__init__.py)

Source: https://context7_llms

This file serves as the main entry point for the 'reactivity' package, exposing core functionalities to users. It imports and re-exports various reactive programming primitives like signals, effects, and memoization functions, as well as collection utilities and context management.

```python
from ._curried import async_derived, async_effect, batch, derived, derived_method, derived_property, effect, memoized, memoized_method, memoized_property, signal, state
from .collections import reactive
from .context import new_context

__all__ = [
    "async_derived",
    "async_effect",
    "batch",
    "derived",
    "derived_method",
    "derived_property",
    "effect",
    "memoized",
    "memoized_method",
    "memoized_property",
    "new_context",
    "reactive",
    "signal",
    "state",
]

```

--------------------------------

### Signal and Effect Management with Warnings in Python

Source: https://context7_llms

Shows how to use `Signal`, `cache`, `effect`, and `warns` from the `reactivity` library. It demonstrates tracking dependencies, emitting warnings on state changes, and ensuring effects are correctly managed. The code relies on `Path` for file path operations.

```python
def test_no_longer_reactive_warning():
    s = Signal(0)

    @cache
    def f():
        return s.get()

    with capture_stdout() as stdout:

        @effect
        def g():
            print(f())

        assert stdout.delta == "0\n"
        assert s.subscribers == {g}

        with warns(RuntimeWarning) as record:
            s.set(1)

    assert stdout.delta == "0\n"
    [warning] = record.list
    assert Path(warning.filename) == Path(__file__)
    assert not g.dependencies
```

--------------------------------

### Reactive Expression with Potential Glitch

Source: https://wikipedia.org/wiki/Reactive_programming

Illustrates a reactive expression where evaluation order can lead to temporary inconsistencies ('glitches'). The expression `g = (t > seconds)` should always be true if `t = seconds + 1`, but evaluation order can cause it to momentarily evaluate to false.

```Conceptual
t = seconds + 1
g = (t > seconds)
```

--------------------------------

### Reactivity Loss and Restore Strategy in Python

Source: https://context7_llms

Demonstrates how to manage reactivity loss in derived properties using the `reactivity_loss_strategy` attribute. It shows setting the strategy to 'restore' or 'ignore' and how it affects dependency tracking and property updates based on conditional logic.

```python
def test_reactivity_loss_strategy():
    s = Signal(1)
    trivial_condition = True
    reactive_condition = Signal(True)

    @Derived
    def f():
        if trivial_condition and reactive_condition.get():
            return s.get()

    assert f() == 1

    f.reactivity_loss_strategy = "restore"
    trivial_condition = False
    reactive_condition.set(False)
    assert f() is None
    assert f.dependencies  # lost but restored
    s.set(2)
    trivial_condition = True
    assert f() is None
    reactive_condition.set(True)
    assert f() == 2

    f.reactivity_loss_strategy = "ignore"
    trivial_condition = False
    reactive_condition.set(False)
    assert f() is None
    assert not f.dependencies  # not restored
    s.set(3)
    trivial_condition = True
    reactive_condition.set(True)
    assert f() is None
```

--------------------------------

### Getting Items from ReactiveSequenceProxy in Python

Source: https://context7_llms

Implements the __getitem__ method for ReactiveSequenceProxy. It handles both integer indices and slices. For slices, it tracks relevant keys and iterators, potentially returning a weakly derived value for efficiency and correctness, especially when equality checks are enabled.

```Python
@overload
    def __getitem__(self, key: int) -> T: ...
    @overload
    def __getitem__(self, key: slice) -> list[T]: ...

    def __getitem__(self, key: int | slice):
        if isinstance(key, slice):
            start, stop, step = key.indices(self._length)
            if step != 1:
                raise NotImplementedError  # TODO
            for i in range(start, stop):
                self._keys[i].track()
            if not self._check_equality:
                self._iter.track()
                return self._data[start:stop]
            # The following implementation is inefficient but works. TODO: refactor this
            return _weak_derived(lambda: (self._iter.track(), self._data[slice(*key.indices(self._length))])[1])()

        else:
            # Handle integer indices
            self._keys[key].track()
            if -self._length <= key < self._length:
                return self._data[key]
            raise IndexError(key)
```

--------------------------------

### Reactivity Primitives and Helpers

Source: https://context7_llms

Imports various components from the 'reactivity' library, including core primitives like `Reactive`, `Signal`, `Effect`, and helpers for memoization and context management. This forms the basis for testing reactive programming patterns.

```python
import gc
from functools import cache
from inspect import ismethod
from pathlib import Path
from typing import assert_type
from warnings import filterwarnings
from weakref import finalize

from pytest import WarningsRecorder, raises, warns
from reactivity import Reactive, batch, create_signal, effect, memoized, memoized_method, memoized_property
from reactivity.context import default_context, new_context
from reactivity.helpers import DerivedProperty, MemoizedMethod, MemoizedProperty
from reactivity.hmr.proxy import Proxy
from reactivity.primitives import Derived, Effect, Signal, State
```

--------------------------------

### Testing CLI File Entry Point in Python

Source: https://context7_llms

This snippet tests the command-line interface (CLI) functionality for running a specific Python file as an entry point. It uses a mock reloader and environment manipulation to simulate the execution of `python a/b.py` and verifies that the script runs successfully, producing the expected output and an exit code of zero.

```python
def test_entry_file():
    with environment() as env, mock_reloader():
        env["a/b.py"] = "if __name__ == '__main__': print(123)"
        assert cli(["a/b.py"]) == 0
        assert env.stdout_delta == "123\n"
```

--------------------------------

### Check for Pulled Dependencies in Python

Source: https://context7_llms

The `_pulled` function determines if a subscribable object has any direct or indirect subscribers that are not `BaseDerived` instances. This is used to decide whether to trigger updates. It performs a breadth-first search (BFS) starting from the given subscribable to check its subscriber graph.

```python
def _pulled(sub: Subscribable):
    visited = set()
    to_visit: set[Subscribable] = {sub}
    while to_visit:
        visited.add(current := to_visit.pop())
        for s in current.subscribers:
            if not isinstance(s, BaseDerived):
                return True
            if s not in visited:
                to_visit.add(s)
    return False
```

--------------------------------

### Python __init__ for utils package

Source: https://context7_llms

Initializes the 'utils' package by importing key components from submodules. It defines the '__all__' attribute to control the public interface of the package. No external dependencies beyond standard library.

```Python
from .io import capture_stdout
from .lineno import current_lineno
from .time import Clock
from .tmpenv import environment
from .trio import create_trio_task_factory, run_trio_in_asyncio

__all__ = "Clock", "capture_stdout", "create_trio_task_factory", "current_lineno", "environment", "run_trio_in_asyncio"
```

--------------------------------

### Implement Signal Class (Python)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

The `Signal` class is a core component for reactive programming, allowing values to be tracked and updated. It inherits from `Subscribable` and includes methods for getting, setting, and updating the signal's value. The `check_equality` parameter allows to control update behavior.

```python
class Signal[T](Subscribable):
    def __init__(self, initial_value: T = None, check_equality=True, *, context: Context | None = None):
        super().__init__(context=context)
        self._value: T = initial_value
        self._check_equality = check_equality

    def get(self, track=True):
        if track:
            self.track()
        return self._value

    def set(self, value: T):
        if not self._check_equality or not _equal(self._value, value):
            self._value = value
            self.notify()
            return True
        return False

    def update(self, updater: Callable[[T], T]):
        return self.set(updater(self._value))
```

--------------------------------

### Reactive Object Initialization and Iteration

Source: https://context7_llms

Covers the behavior of the `Reactive` object during initialization and iteration. It demonstrates that accessing non-existent keys raises a `KeyError` and that iterating over an empty `Reactive` object yields an empty result.

```python
def test_reactive_spread():
    obj = Reactive()
    with raises(KeyError, match="key"):
        obj["key"]

    assert {**obj} == {{}}
    assert len(obj) == 0
```

--------------------------------

### HMR Package Initialization with reactivity/__init__.py

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

Initializes the HMR Python package by importing and re-exporting core components from internal modules. This file makes reactive programming primitives like signals, effects, and state, as well as collection and context management, directly accessible from the top-level 'reactivity' package.

```python
from ._curried import async_derived, async_effect, batch, derived, derived_method, derived_property, effect, memoized, memoized_method, memoized_property, signal, state
from .collections import reactive
from .context import new_context

__all__ = [
    "async_derived",
    "async_effect",
    "batch",
    "derived",
    "derived_method",
    "derived_property",
    "effect",
    "memoized",
    "memoized_method",
    "memoized_property",
    "new_context",
    "reactive",
    "signal",
    "state",
]
```

--------------------------------

### Python Project Configuration (pyproject.toml)

Source: https://context7_llms

Defines the project's metadata, dependencies, build system, and script entry points. It specifies project name, versioning, classifiers, keywords, Python version requirements, and external dependencies like 'watchfiles'. The tool.pdm section configures the build backend and versioning source.

```toml
[project]
name = "hmr"
readme = "README.md"
dynamic = ["version"]
classifiers = [
    "Development Status :: 5 - Production/Stable",
    "Intended Audience :: Developers",
    "Operating System :: OS Independent",
    "Topic :: Software Development :: Libraries :: Python Modules",
    "Typing :: Typed",
]
keywords = ["reactive-programming", "hot-module-reload", "signals", "hmr"]
requires-python = ">=3.12"
description = "Hot Module Reload and Fine-grained Reactive Programming for Python"
dependencies = ["watchfiles>=0.21,<2 ; sys_platform != 'emscripten'"]

[project.scripts]
hmr = "reactivity.hmr.run:main"

[project.urls]
Homepage = "https://pyth-on-line.promplate.dev/hmr"
Documentation = "https://hmr.promplate.dev/"
Repository = "https://github.com/promplate/hmr"
Changelog = "https://github.com/promplate/pyth-on-line/commits/main/packages/hmr"

[build-system]
requires = ["pdm-backend"]
build-backend = "pdm.backend"

[tool.pdm]
version = { source = "file", path = "reactivity/hmr/core.py" }

```

--------------------------------

### Python HMR Package Initialization (__init__.py)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

This Python script serves as the main entry point for the HMR package, defining the public API by importing specific modules and functions. It makes `cache_across_reloads`, `cli`, `on_dispose`, `post_reload`, and `pre_reload` available for use when the package is imported.

```python
from .hooks import on_dispose, post_reload, pre_reload
from .run import cli
from .utils import cache_across_reloads

__all__ = ("cache_across_reloads", "cli", "on_dispose", "post_reload", "pre_reload")
```

--------------------------------

### Install HMR Docs MCP Server via HTTP

Source: https://pyth-on-line.promplate.dev/hmr/mcp

This configuration allows you to access the HMR Docs MCP Server using the Streamable HTTP protocol. It's the recommended and latest method for integrating with MCP clients.

```json
{
  "type": "http",
  "url": "https://pyth-on-line.promplate.dev/hmr/mcp"
}
```

--------------------------------

### Context Management for Reactive State in Python

Source: https://context7_llms

Demonstrates how to manage reactive state across different contexts using `new_context()`. It shows that effects and state can be scoped to specific contexts, allowing for isolated reactive updates. This is useful for building complex applications with independent reactive components.

```python
def test_context():
    a = new_context()
    b = new_context()

    class Rect:
        x = State(1, context=a)
        y = State(2, context=b)

        @property
        def size(self):
            return self.x * self.y

    r = Rect()

    with capture_stdout() as stdout, a.effect(lambda: print(f"a{r.size}"), context=a), b.effect(lambda: print(f"b{r.size}"), context=b):
        assert stdout.delta == "a2\nb2\n"
        r.x = 3
        assert stdout.delta == "a6\n"
        r.y = 4
        assert stdout.delta == "b12\n"

def test_context_usage_with_reactive_namespace():
    c = new_context()
    dct = Reactive(context=c)

    with capture_stdout() as stdout:

        @effect(context=c)
        def _():
            try:
                print(dct[1])
            except KeyError:
                print()

        assert stdout.delta == "\n"
        dct[1] = 2
        assert stdout.delta == "2\n"
```

--------------------------------

### SyncReloader for Synchronous File Watching

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

The SyncReloader class extends BaseReloader to implement synchronous file watching using the 'watchfiles' library. It provides methods to start watching files and to keep watching until an interrupt signal is received. It integrates with HMR_CONTEXT for effects and handles KeyboardInterrupt exceptions.

```python
class SyncReloader(BaseReloader):
    def start_watching(self):
        from watchfiles import watch

        for events in watch(self.entry, *self.includes, stop_event=self._stop_event):
            self.on_events(events)

        del self._stop_event

    def keep_watching_until_interrupt(self):
        call_pre_reload_hooks()
        with suppress(KeyboardInterrupt), HMR_CONTEXT.effect(self.run_entry_file):
            call_post_reload_hooks()
            self.start_watching()

```

--------------------------------

### watchfiles CLI Help Output

Source: https://watchfiles.helpmanual.io/cli

This snippet displays the help message for the watchfiles CLI, outlining its usage, arguments, and available options. It details how to specify target commands or functions, directories to watch, file filters, and various configuration settings for process management and logging.

```bash
usage: watchfiles [-h] [--ignore-paths [IGNORE_PATHS]] [--target-type [{command,function,auto}]] [--filter [FILTER]] [--args [ARGS]] [--verbose] [--non-recursive] [--verbosity [{warning,info,debug}]] [--sigint-timeout [SIGINT_TIMEOUT]] [--grace-period [GRACE_PERIOD]] [--sigkill-timeout [SIGKILL_TIMEOUT]] [--ignore-permission-denied] [--version]
target [paths ...]

Watch one or more directories and execute either a shell command or a python function on file changes.

Example of watching the current directory and calling a python function: watchfiles foobar.main
Example of watching python files in two local directories and calling a shell command: watchfiles --filter python 'pytest --lf' src tests

See https://watchfiles.helpmanual.io/cli/ for more information.

positional arguments:
  target                Command or dotted function path to run
  paths                 Filesystem paths to watch, defaults to current directory

options:
  -h, --help            show this help message and exit
  --ignore-paths [IGNORE_PATHS]
                        Specify directories to ignore, to ignore multiple paths use a comma as separator, e.g. "env" or "env,node_modules"
  --target-type [{command,function,auto}]
                        Whether the target should be intercepted as a shell command or a python function, defaults to "auto" which infers the target type from the target string
  --filter [FILTER]     Which files to watch, defaults to "default" which uses the "DefaultFilter", "python" uses the "PythonFilter", "all" uses no filter, any other value is interpreted as a python function/class path which is imported
  --args [ARGS]         Arguments to set on sys.argv before calling target function, used only if the target is a function
  --verbose             Set log level to "debug", wins over `--verbosity`
  --non-recursive       Do not watch for changes in sub-directories recursively
  --verbosity [{warning,info,debug}]
                        Log level, defaults to "info"
  --sigint-timeout [SIGINT_TIMEOUT]
                        How long to wait for the sigint timeout before sending sigkill.
  --grace-period [GRACE_PERIOD]
                        Number of seconds after the process is started before watching for changes.
  --sigkill-timeout [SIGKILL_TIMEOUT]
                        How long to wait for the sigkill timeout before issuing a timeout exception.
  --ignore-permission-denied
                        Ignore permission denied errors while watching files and directories.
  --version, -V         show program's version number and exit
```

--------------------------------

### HMR Hooks for Pre and Post Reload Operations (Python)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

This module defines functions and decorators for managing pre- and post-hot module replacement hooks. It allows registration of callbacks to be executed before or after module reloads, facilitating cleanup or setup tasks. It includes context managers for temporary hook registration.

```python
from collections.abc import Callable
from contextlib import contextmanager
from inspect import currentframe
from pathlib import Path
from typing import Any

pre_reload_hooks: dict[str, Callable[[], Any]] = {}
post_reload_hooks: dict[str, Callable[[], Any]] = {}


def pre_reload[T](func: Callable[[], T]) -> Callable[[], T]:
    pre_reload_hooks[func.__name__] = func
    return func


def post_reload[T](func: Callable[[], T]) -> Callable[[], T]:
    post_reload_hooks[func.__name__] = func
    return func


@contextmanager
def use_pre_reload(func):
    pre_reload(func)
    try:
        yield func
    finally:
        pre_reload_hooks.pop(func.__name__, None)


@contextmanager
def use_post_reload(func):
    post_reload(func)
    try:
        yield func
    finally:
        post_reload_hooks.pop(func.__name__, None)


def call_pre_reload_hooks():
    for func in pre_reload_hooks.values():
        func()


def call_post_reload_hooks():
    for func in post_reload_hooks.values():
        func()


def on_dispose(func: Callable[[], Any], __file__: str | None = None):
    path = Path(currentframe().f_back.f_globals["__file__"] if __file__ is None else __file__).resolve()  # type: ignore

    from .core import ReactiveModule

    module = ReactiveModule.instances[path]
    module.register_dispose_callback(func)

```

--------------------------------

### Python Class Initialization with Sequence and Equality Check

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

Initializes a Python class instance. Accepts an optional sequence for initial data, a boolean to control equality checks, and an optional context object. It calls the parent class's constructor with the processed initial data.

```python
def __init__(self, initial: Sequence[T] | None = None, check_equality=True, *, context: Context | None = None):
    super().__init__([*initial] if initial is not None else [], check_equality, context=context)
```

--------------------------------

### Install HMR Docs MCP Server via stdio (npx mcp-remote)

Source: https://pyth-on-line.promplate.dev/hmr/mcp

This configuration uses the 'mcp-remote' command-line tool via npx to proxy the connection to the HMR Docs MCP Server. This method is useful for command-line or specific client integrations.

```json
{
  "type": "stdio",
  "command": "npx",
  "args": [
    "mcp-remote",
    "https://pyth-on-line.promplate.dev/hmr/mcp"
  ]
}
```

--------------------------------

### Python: Dynamically Define Slots and Error Handling

Source: https://context7_llms

This code snippet focuses on dynamically defining slots for a class and raising a TypeError if a required slot is missing. It constructs an informative error message that guides the user on how to update the __slots__ definition. The code uses inspect to extract class definition details.

```python
key = f"{self.__class__.__name__}.SLOT_KEY"
match slots:
    case tuple() as slots:
        new_slots = f"({', '.join(slots)}, {key})" if slots else f"({key},)"
    case str():
        new_slots = f"{slots}, {key}"
    case set():
        new_slots = f"{{{', '.join(slots)}, {key}}}" if slots else f"{{{key}}}"
    case _:
        new_slots = f"[{', '.join(slots)}, {key}]" if slots else f"[{key}]"

from inspect import getsource
from textwrap import dedent, indent

try:
    selected = []
    for line in dedent(getsource(owner)).splitlines():
        if line.startswith(("@", f"class {owner.__name__}")):
            selected.append(line)
        else:
            break
    cls_def = "\n".join(selected)
    # maybe source mismatch (usually during `exec`)
    if f"class {owner.__name__}" not in selected:
        raise OSError  # noqa: TRY301
except (OSError, TypeError):
    bases = [b.__name__ for b in owner.__bases__ if b is not object]
    cls_def = f"class {owner.__name__}{f'({", ".join(bases)})' if bases else ''}:"

__tracebackhide__ = 1  # for pytest

msg = f"Missing {key} in slots definition for \`{self.__class__.__name__}\`.\n\n"
msg += indent(
    "\n\n".join(
        (
            f"Please add \`{key}\` to your \`__slots__`. You should change:",
            indent(f"{cls_def}\n    __slots__ = {slots!r}", "  "),
            "to:",
            indent(f"{cls_def}\n    __slots__ = {new_slots}", "  "),
        )
    ),
    "  ",
)
raise TypeError(msg + "\n")
```

--------------------------------

### Testing CLI Module Entry Point in Python

Source: https://context7_llms

This snippet tests the command-line interface (CLI) functionality for running a Python module as an entry point. It uses a mock reloader and environment manipulation to simulate the execution of `python -m a.b` and asserts that the correct output is produced and the exit code is zero. It also tests running a directory as an entry point.

```python
def test_entry_module():
    with environment() as env, mock_reloader():
        env["a/b/__init__.py"].touch()
        env["a/b/__main__.py"] = "if __name__ == '__main__': print(123)"
        assert cli(["-m", "a.b"]) == 0
        assert env.stdout_delta == "123\n"
        assert cli(["a/b"]) == 0
        assert env.stdout_delta == "123\n"
```

--------------------------------

### Batch Updates with Effects

Source: https://context7_llms

Demonstrates how to group multiple state changes into a single update cycle using the `batch` context manager. This prevents intermediate re-computations of effects and derived states until all changes within the batch are applied.

```python
def test_batch():
    class Example:
        value = State(0)

    obj = Example()

    history = []

    @effect
    def _():
        history.append(obj.value)

    assert history == [0]

    def increment():
        obj.value += 1

    increment()
    assert history == [0, 1]

    increment()
    increment()
    assert history == [0, 1, 2, 3]

    with batch():
        increment()
        increment()
        assert history == [0, 1, 2, 3]

    assert history == [0, 1, 2, 3, 5]
```

--------------------------------

### Run Shell Command with watchfiles CLI

Source: https://watchfiles.helpmanual.io/cli

This example shows how to use the watchfiles CLI to execute a shell command, like running tests, and automatically re-run it when files change. It utilizes pytest's '--lf' (last-failed) option for efficiency. The CLI watches the current directory and its subdirectories by default.

```bash
watchfiles 'pytest --lf'
```

--------------------------------

### Python HMR API: Sync Reloader (api.py)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

This Python code defines the `SyncReloaderAPI` class, which integrates with `SyncReloader` and `LifecycleMixin` to provide a synchronous interface for hot module reloading. It includes context management methods (`__enter__`, `__exit__`) to start, stop, and manage the HMR watching thread, along with hooks for pre- and post-reload actions.

```python
import sys
from .core import HMR_CONTEXT, AsyncReloader, BaseReloader, SyncReloader
from .hooks import call_post_reload_hooks, call_pre_reload_hooks


class LifecycleMixin(BaseReloader):
    def run_with_hooks(self):
        self._original_main_module = sys.modules["__main__"]
        sys.modules["__main__"] = self.entry_module
        call_pre_reload_hooks()
        self.effect = HMR_CONTEXT.effect(self.run_entry_file)
        call_post_reload_hooks()

    def clean_up(self):
        self.effect.dispose()
        self.entry_module.load.dispose()
        self.entry_module.load.invalidate()
        sys.modules["__main__"] = self._original_main_module


class SyncReloaderAPI(SyncReloader, LifecycleMixin):
    def __enter__(self):
        from threading import Thread

        self.run_with_hooks()
        self.thread = Thread(target=self.start_watching)
        self.thread.start()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(to_thread(self.start_watching))
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()


class AsyncReloaderAPI(AsyncReloader, LifecycleMixin):
    def __enter__(self):
        from asyncio import run
        from threading import Event, Thread

        self.run_with_hooks()

        e = Event()

        async def task():
            e.set()
            await self.start_watching()

        self.thread = Thread(target=lambda: run(task()))
        self.thread.start()
        e.wait()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(self.start_watching())
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()
```

--------------------------------

### Context Management for Reactive Computations in Python

Source: https://context7_llms

Defines a `Context` object for managing reactive programming state, including current computations and batch scheduling. The `enter` method provides a context manager for managing the lifecycle of a computation, ensuring proper handling of dependencies and potential exceptions during execution. It also includes logic for handling reactivity loss strategies.

```python
class Context(NamedTuple):
    current_computations: list[BaseComputation]
    batches: list[Batch]
    async_execution_context: ContextVar[Context | None]

    def schedule_callbacks(self, callbacks: Iterable[BaseComputation]):
        self.batches[-1].callbacks.update(callbacks)

    @contextmanager
    def enter(self, computation: BaseComputation):
        old_dependencies = {*computation.dependencies}
        computation.dispose()
        self.current_computations.append(computation)
        try:
            yield
        except BaseException:
            # For backward compatibility, we restore old dependencies only if some dependencies are lost after an exception.
            # This behavior may be configurable in the future.
            if computation.dependencies.issubset(old_dependencies):
                for dep in old_dependencies:
                    dep.subscribers.add(computation)
                computation.dependencies.update(old_dependencies)
            raise
        else:
            if not computation.dependencies and (strategy := computation.reactivity_loss_strategy) != "ignore":
                if strategy == "restore" and old_dependencies:
                    for dep in old_dependencies:
                        dep.subscribers.add(computation)
                    computation.dependencies.update(old_dependencies)
                    return
                from pathlib import Path
                from sysconfig import get_path
                from warnings import warn

                msg = "lost all its dependencies" if old_dependencies else "has no dependencies"
                warn(f"{computation} {msg} and will never be auto-triggered.", RuntimeWarning, skip_file_prefixes=(str(Path(__file__).parent), str(Path(get_path("stdlib")).resolve())))
        finally:
            last = self.current_computations.pop()
            assert last is computation  # sanity check

    @property
    def batch(self):
        return partial(Batch, context=self)

    @property
    def signal(self):
        return partial(Signal, context=self)

    @property
    def effect(self):
```

--------------------------------

### Watch Custom Directories and Files with watchfiles CLI

Source: https://watchfiles.helpmanual.io/cli

This example demonstrates how to configure the watchfiles CLI to monitor specific directories and react only to changes in certain file types. It watches the 'src' and 'tests' directories and triggers the 'pytest --lf' command only when Python files are modified. The '--filter python' option specifies that only Python files should be watched.

```bash
watchfiles --filter python 'pytest --lf' src tests
```

--------------------------------

### Python Reactivity Context Management (`reactivity/context.py`)

Source: https://pyth-on-line.promplate.dev/hmr/llms.txt

Manages the reactivity context in Python, including current computations, batching, and asynchronous execution. It provides methods to enter and exit computation contexts, handle dependency tracking, and manage warnings for lost dependencies. This context is crucial for the internal workings of the reactivity system.

```python
from __future__ import annotations

from collections.abc import Iterable
from contextlib import contextmanager
from contextvars import ContextVar
from functools import partial
from typing import TYPE_CHECKING, NamedTuple

if TYPE_CHECKING:
    from .primitives import BaseComputation


class Context(NamedTuple):
    current_computations: list[BaseComputation]
    batches: list[Batch]
    async_execution_context: ContextVar[Context | None]

    def schedule_callbacks(self, callbacks: Iterable[BaseComputation]):
        self.batches[-1].callbacks.update(callbacks)

    @contextmanager
    def enter(self, computation: BaseComputation):
        old_dependencies = {*computation.dependencies}
        computation.dispose()
        self.current_computations.append(computation)
        try:
            yield
        except BaseException:
            if computation.dependencies.issubset(old_dependencies):
                for dep in old_dependencies:
                    dep.subscribers.add(computation)
                computation.dependencies.update(old_dependencies)
            raise
        else:
            if not computation.dependencies and (strategy := computation.reactivity_loss_strategy) != "ignore":
                if strategy == "restore" and old_dependencies:
                    for dep in old_dependencies:
                        dep.subscribers.add(computation)
                    computation.dependencies.update(old_dependencies)
                    return
                from pathlib import Path
                from sysconfig import get_path
                from warnings import warn

                msg = "lost all its dependencies" if old_dependencies else "has no dependencies"
                warn(f"{computation} {msg} and will never be auto-triggered.", RuntimeWarning, skip_file_prefixes=(str(Path(__file__).parent), str(Path(get_path("stdlib")).resolve())))
        finally:
            last = self.current_computations.pop()
            assert last is computation

    @property
    def batch(self):
        return partial(Batch, context=self)

    @property
    def signal(self):
        return partial(Signal, context=self)

    @property
    def effect(self):
        return partial(Effect, context=self)

    @property
    def derived(self):
        return partial(Derived, context=self)

    @property
    def async_effect(self):
        return partial(AsyncEffect, context=self)

    @property
    def async_derived(self):
        return partial(AsyncDerived, context=self)

    @contextmanager
    def untrack(self):
        computations = self.current_computations[:]
        self.current_computations.clear()
        try:
            yield
        finally:
            self.current_computations[:] = computations

    @property
    def leaf(self):
        return self.async_execution_context.get() or self

    def fork(self):
        self.async_execution_context.set(Context(self.current_computations[:], self.batches[:], self.async_execution_context))


def new_context():
    return Context([], [], async_execution_context=ContextVar("current context", default=None))

default_context = new_context()

from .async_primitives import AsyncDerived, AsyncEffect
from .primitives import Batch, Derived, Effect, Signal

```

--------------------------------

### Create and Use Signals with Derived States

Source: https://context7_llms

Demonstrates creating signals, defining derived states that depend on signals, and observing their behavior with state updates and invalidations. It shows how `Derived` functions re-execute when their dependencies change and how `invalidate` can force re-computation.

```python
def test_derived():
    get_s, set_s = create_signal(0)

    @Derived
    def f():
        print(get_s())
        return get_s() + 1

    with capture_stdout() as stdout:
        assert stdout == ""
        assert f() == 1
        assert stdout == "0\n"
        f()
        assert stdout == "0\n"
        set_s(1)
        assert stdout == "0\n"
        assert f() == 2
        assert stdout == "0\n1\n"
        set_s(1)
        f()
        assert stdout == "0\n1\n"

    @Derived
    def g():
        print(f() + 1)
        return f() + 1

    with capture_stdout() as stdout:
        assert g() == 3
        assert stdout.delta == "3\n"
        f.invalidate()
        assert stdout.delta == ""
        assert g() == 3
        assert stdout.delta == "1\n"
```

--------------------------------

### Python File System Utilities (FsUtils)

Source: https://context7_llms

Provides utilities for file system operations including writing, replacing content, and touching files. It supports a dictionary-like interface for direct file content manipulation and uses pathlib for path operations. Dependencies include 'pathlib' and 'linecache'.

```Python
from functools import partial
from linecache import cache
from pathlib import Path
from textwrap import dedent
from typing import final


class FsUtils:
    def write(self, filepath: str, content: str):
        path = Path(filepath)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content)
        cache.pop(filepath, None)

    def replace(self, filepath: str, old: str, new: str):
        path = Path(filepath)
        path.write_text(path.read_text().replace(old, new))
        cache.pop(filepath, None)

    def touch(self, filepath: str):
        path = Path(filepath)
        self.write(filepath, path.read_text() if path.exists() else "")

    @final
    def __getitem__(self, filepath: str):
        class Replacer:
            replace = staticmethod(partial(self.replace, filepath))
            touch = staticmethod(partial(self.touch, filepath))

        return Replacer()

    @final
    def __setitem__(self, filepath: str, content: str):
        self.write(filepath, dedent(content))
```

--------------------------------

### Initialize HMR Context and Utilities in Python

Source: https://context7_llms

This module sets up the global HMR context using `new_context` and imports necessary utilities like `cache_across_reloads` and CLI functions. It defines the `__all__` list to control the public API of the `hmr` package.

```python
from ..context import new_context

HMR_CONTEXT = new_context()

from .hooks import on_dispose, post_reload, pre_reload
from .run import cli
from .utils import cache_across_reloads

__all__ = ("cache_across_reloads", "cli", "on_dispose", "post_reload", "pre_reload")
```