We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Remove `Reactive.var` in favour of `var`?
In `reactive.py` we have the classmethod `Reactive.var` and the class `var` that subclasses `Reactive`.
They are not strictly equivalent, but they seem to have the same purpose.
Should we delete `Reactive.var`?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/textual/reactive.py`
Content:
```
1 """
2 
3 The `Reactive` class implements [reactivity](/guide/reactivity/).
4 """
5 
6 from __future__ import annotations
7 
8 from functools import partial
9 from inspect import isawaitable
10 from typing import (
11     TYPE_CHECKING,
12     Any,
13     Awaitable,
14     Callable,
15     ClassVar,
16     Generic,
17     Type,
18     TypeVar,
19 )
20 
21 import rich.repr
22 
23 from . import events
24 from ._callback import count_parameters
25 from ._types import MessageTarget, WatchCallbackType
26 
27 if TYPE_CHECKING:
28     from .dom import DOMNode
29 
30     Reactable = DOMNode
31 
32 ReactiveType = TypeVar("ReactiveType")
33 
34 
35 @rich.repr.auto
36 class Reactive(Generic[ReactiveType]):
37     """Reactive descriptor.
38 
39     Args:
40         default: A default value or callable that returns a default.
41         layout: Perform a layout on change.
42         repaint: Perform a repaint on change.
43         init: Call watchers on initialize (post mount).
44         always_update: Call watchers even when the new value equals the old value.
45         compute: Run compute methods when attribute is changed.
46     """
47 
48     _reactives: ClassVar[dict[str, object]] = {}
49 
50     def __init__(
51         self,
52         default: ReactiveType | Callable[[], ReactiveType],
53         *,
54         layout: bool = False,
55         repaint: bool = True,
56         init: bool = False,
57         always_update: bool = False,
58         compute: bool = True,
59     ) -> None:
60         self._default = default
61         self._layout = layout
62         self._repaint = repaint
63         self._init = init
64         self._always_update = always_update
65         self._run_compute = compute
66 
67     def __rich_repr__(self) -> rich.repr.Result:
68         yield self._default
69         yield "layout", self._layout
70         yield "repaint", self._repaint
71         yield "init", self._init
72         yield "always_update", self._always_update
73         yield "compute", self._run_compute
74 
75     @classmethod
76     def var(
77         cls,
78         default: ReactiveType | Callable[[], ReactiveType],
79         always_update: bool = False,
80     ) -> Reactive:
81         """A reactive variable that doesn't update or layout.
82 
83         Args:
84             default: A default value or callable that returns a default.
85             always_update: Call watchers even when the new value equals the old value.
86 
87         Returns:
88             A Reactive descriptor.
89         """
90         return cls(default, layout=False, repaint=False, init=False)
91 
92     def _initialize_reactive(self, obj: Reactable, name: str) -> None:
93         """Initialized a reactive attribute on an object.
94 
95         Args:
96             obj: An object with reactive attributes.
97             name: Name of attribute.
98         """
99         _rich_traceback_omit = True
100         internal_name = f"_reactive_{name}"
101         if hasattr(obj, internal_name):
102             # Attribute already has a value
103             return
104 
105         compute_method = getattr(obj, f"compute_{name}", None)
106         if compute_method is not None and self._init:
107             default = getattr(obj, f"compute_{name}")()
108         else:
109             default_or_callable = self._default
110             default = (
111                 default_or_callable()
112                 if callable(default_or_callable)
113                 else default_or_callable
114             )
115         setattr(obj, internal_name, default)
116         if self._init:
117             self._check_watchers(obj, name, default)
118 
119     @classmethod
120     def _initialize_object(cls, obj: Reactable) -> None:
121         """Set defaults and call any watchers / computes for the first time.
122 
123         Args:
124             obj: An object with Reactive descriptors
125         """
126         _rich_traceback_omit = True
127         for name, reactive in obj._reactives.items():
128             reactive._initialize_reactive(obj, name)
129 
130     @classmethod
131     def _reset_object(cls, obj: object) -> None:
132         """Reset reactive structures on object (to avoid reference cycles).
133 
134         Args:
135             obj: A reactive object.
136         """
137         getattr(obj, "__watchers", {}).clear()
138         getattr(obj, "__computes", []).clear()
139 
140     def __set_name__(self, owner: Type[MessageTarget], name: str) -> None:
141         # Check for compute method
142         if hasattr(owner, f"compute_{name}"):
143             # Compute methods are stored in a list called `__computes`
144             try:
145                 computes = getattr(owner, "__computes")
146             except AttributeError:
147                 computes = []
148                 setattr(owner, "__computes", computes)
149             computes.append(name)
150 
151         # The name of the attribute
152         self.name = name
153         # The internal name where the attribute's value is stored
154         self.internal_name = f"_reactive_{name}"
155         self.compute_name = f"compute_{name}"
156         default = self._default
157         setattr(owner, f"_default_{name}", default)
158 
159     def __get__(self, obj: Reactable, obj_type: type[object]) -> ReactiveType:
160         internal_name = self.internal_name
161         if not hasattr(obj, internal_name):
162             self._initialize_reactive(obj, self.name)
163 
164         if hasattr(obj, self.compute_name):
165             value: ReactiveType
166             old_value = getattr(obj, internal_name)
167             _rich_traceback_omit = True
168             value = getattr(obj, self.compute_name)()
169             setattr(obj, internal_name, value)
170             self._check_watchers(obj, self.name, old_value)
171             return value
172         else:
173             return getattr(obj, internal_name)
174 
175     def __set__(self, obj: Reactable, value: ReactiveType) -> None:
176         _rich_traceback_omit = True
177 
178         self._initialize_reactive(obj, self.name)
179 
180         if hasattr(obj, self.compute_name):
181             raise AttributeError(
182                 f"Can't set {obj}.{self.name!r}; reactive attributes with a compute method are read-only"
183             )
184 
185         name = self.name
186         current_value = getattr(obj, name)
187         # Check for validate function
188         validate_function = getattr(obj, f"validate_{name}", None)
189         # Call validate
190         if callable(validate_function):
191             value = validate_function(value)
192         # If the value has changed, or this is the first time setting the value
193         if current_value != value or self._always_update:
194             # Store the internal value
195             setattr(obj, self.internal_name, value)
196 
197             # Check all watchers
198             self._check_watchers(obj, name, current_value)
199 
200             if self._run_compute:
201                 self._compute(obj)
202 
203             # Refresh according to descriptor flags
204             if self._layout or self._repaint:
205                 obj.refresh(repaint=self._repaint, layout=self._layout)
206 
207     @classmethod
208     def _check_watchers(cls, obj: Reactable, name: str, old_value: Any):
209         """Check watchers, and call watch methods / computes
210 
211         Args:
212             obj: The reactable object.
213             name: Attribute name.
214             old_value: The old (previous) value of the attribute.
215         """
216         _rich_traceback_omit = True
217         # Get the current value.
218         internal_name = f"_reactive_{name}"
219         value = getattr(obj, internal_name)
220 
221         async def await_watcher(awaitable: Awaitable) -> None:
222             """Coroutine to await an awaitable returned from a watcher"""
223             _rich_traceback_omit = True
224             await awaitable
225             # Watcher may have changed the state, so run compute again
226             obj.post_message(events.Callback(callback=partial(Reactive._compute, obj)))
227 
228         def invoke_watcher(
229             watch_function: Callable, old_value: object, value: object
230         ) -> None:
231             """Invoke a watch function.
232 
233             Args:
234                 watch_function: A watch function, which may be sync or async.
235                 old_value: The old value of the attribute.
236                 value: The new value of the attribute.
237             """
238             _rich_traceback_omit = True
239             param_count = count_parameters(watch_function)
240             if param_count == 2:
241                 watch_result = watch_function(old_value, value)
242             elif param_count == 1:
243                 watch_result = watch_function(value)
244             else:
245                 watch_result = watch_function()
246             if isawaitable(watch_result):
247                 # Result is awaitable, so we need to await it within an async context
248                 obj.post_message(
249                     events.Callback(callback=partial(await_watcher, watch_result))
250                 )
251 
252         private_watch_function = getattr(obj, f"_watch_{name}", None)
253         if callable(private_watch_function):
254             invoke_watcher(private_watch_function, old_value, value)
255 
256         public_watch_function = getattr(obj, f"watch_{name}", None)
257         if callable(public_watch_function):
258             invoke_watcher(public_watch_function, old_value, value)
259 
260         # Process "global" watchers
261         watchers: list[tuple[Reactable, Callable]]
262         watchers = getattr(obj, "__watchers", {}).get(name, [])
263         # Remove any watchers for reactables that have since closed
264         if watchers:
265             watchers[:] = [
266                 (reactable, callback)
267                 for reactable, callback in watchers
268                 if reactable.is_attached and not reactable._closing
269             ]
270             for reactable, callback in watchers:
271                 with reactable.prevent(*obj._prevent_message_types_stack[-1]):
272                     invoke_watcher(callback, old_value, value)
273 
274     @classmethod
275     def _compute(cls, obj: Reactable) -> None:
276         """Invoke all computes.
277 
278         Args:
279             obj: Reactable object.
280         """
281         _rich_traceback_guard = True
282         for compute in obj._reactives.keys():
283             try:
284                 compute_method = getattr(obj, f"compute_{compute}")
285             except AttributeError:
286                 continue
287             current_value = getattr(
288                 obj, f"_reactive_{compute}", getattr(obj, f"_default_{compute}", None)
289             )
290             value = compute_method()
291             setattr(obj, f"_reactive_{compute}", value)
292             if value != current_value:
293                 cls._check_watchers(obj, compute, current_value)
294 
295 
296 class reactive(Reactive[ReactiveType]):
297     """Create a reactive attribute.
298 
299     Args:
300         default: A default value or callable that returns a default.
301         layout: Perform a layout on change.
302         repaint: Perform a repaint on change.
303         init: Call watchers on initialize (post mount).
304         always_update: Call watchers even when the new value equals the old value.
305     """
306 
307     def __init__(
308         self,
309         default: ReactiveType | Callable[[], ReactiveType],
310         *,
311         layout: bool = False,
312         repaint: bool = True,
313         init: bool = True,
314         always_update: bool = False,
315     ) -> None:
316         super().__init__(
317             default,
318             layout=layout,
319             repaint=repaint,
320             init=init,
321             always_update=always_update,
322         )
323 
324 
325 class var(Reactive[ReactiveType]):
326     """Create a reactive attribute (with no auto-refresh).
327 
328     Args:
329         default: A default value or callable that returns a default.
330         init: Call watchers on initialize (post mount).
331         always_update: Call watchers even when the new value equals the old value.
332     """
333 
334     def __init__(
335         self,
336         default: ReactiveType | Callable[[], ReactiveType],
337         init: bool = True,
338         always_update: bool = False,
339     ) -> None:
340         super().__init__(
341             default,
342             layout=False,
343             repaint=False,
344             init=init,
345             always_update=always_update,
346         )
347 
348 
349 def _watch(
350     node: DOMNode,
351     obj: Reactable,
352     attribute_name: str,
353     callback: WatchCallbackType,
354     *,
355     init: bool = True,
356 ) -> None:
357     """Watch a reactive variable on an object.
358 
359     Args:
360         obj: The parent object.
361         attribute_name: The attribute to watch.
362         callback: A callable to call when the attribute changes.
363         init: True to call watcher initialization.
364     """
365     if not hasattr(obj, "__watchers"):
366         setattr(obj, "__watchers", {})
367     watchers: dict[str, list[tuple[Reactable, Callable]]] = getattr(obj, "__watchers")
368     watcher_list = watchers.setdefault(attribute_name, [])
369     if callback in watcher_list:
370         return
371     watcher_list.append((node, callback))
372     if init:
373         current_value = getattr(obj, attribute_name, None)
374         Reactive._check_watchers(obj, attribute_name, current_value)
375
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/textual/reactive.py b/src/textual/reactive.py
--- a/src/textual/reactive.py
+++ b/src/textual/reactive.py
@@ -72,23 +72,6 @@
         yield "always_update", self._always_update
         yield "compute", self._run_compute
 
-    @classmethod
-    def var(
-        cls,
-        default: ReactiveType | Callable[[], ReactiveType],
-        always_update: bool = False,
-    ) -> Reactive:
-        """A reactive variable that doesn't update or layout.
-
-        Args:
-            default: A default value or callable that returns a default.
-            always_update: Call watchers even when the new value equals the old value.
-
-        Returns:
-            A Reactive descriptor.
-        """
-        return cls(default, layout=False, repaint=False, init=False)
-
     def _initialize_reactive(self, obj: Reactable, name: str) -> None:
         """Initialized a reactive attribute on an object.

{"golden_diff": "diff --git a/src/textual/reactive.py b/src/textual/reactive.py\n--- a/src/textual/reactive.py\n+++ b/src/textual/reactive.py\n@@ -72,23 +72,6 @@\n         yield \"always_update\", self._always_update\n         yield \"compute\", self._run_compute\n \n-    @classmethod\n-    def var(\n-        cls,\n-        default: ReactiveType | Callable[[], ReactiveType],\n-        always_update: bool = False,\n-    ) -> Reactive:\n-        \"\"\"A reactive variable that doesn't update or layout.\n-\n-        Args:\n-            default: A default value or callable that returns a default.\n-            always_update: Call watchers even when the new value equals the old value.\n-\n-        Returns:\n-            A Reactive descriptor.\n-        \"\"\"\n-        return cls(default, layout=False, repaint=False, init=False)\n-\n     def _initialize_reactive(self, obj: Reactable, name: str) -> None:\n         \"\"\"Initialized a reactive attribute on an object.\n", "issue": "Remove `Reactive.var` in favour of `var`?\nIn `reactive.py` we have the classmethod `Reactive.var` and the class `var` that subclasses `Reactive`.\r\nThey are not strictly equivalent, but they seem to have the same purpose.\r\nShould we delete `Reactive.var`?\n", "before_files": [{"content": "\"\"\"\n\nThe `Reactive` class implements [reactivity](/guide/reactivity/).\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom functools import partial\nfrom inspect import isawaitable\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Awaitable,\n    Callable,\n    ClassVar,\n    Generic,\n    Type,\n    TypeVar,\n)\n\nimport rich.repr\n\nfrom . import events\nfrom ._callback import count_parameters\nfrom ._types import MessageTarget, WatchCallbackType\n\nif TYPE_CHECKING:\n    from .dom import DOMNode\n\n    Reactable = DOMNode\n\nReactiveType = TypeVar(\"ReactiveType\")\n\n\n@rich.repr.auto\nclass Reactive(Generic[ReactiveType]):\n    \"\"\"Reactive descriptor.\n\n    Args:\n        default: A default value or callable that returns a default.\n        layout: Perform a layout on change.\n        repaint: Perform a repaint on change.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n        compute: Run compute methods when attribute is changed.\n    \"\"\"\n\n    _reactives: ClassVar[dict[str, object]] = {}\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        *,\n        layout: bool = False,\n        repaint: bool = True,\n        init: bool = False,\n        always_update: bool = False,\n        compute: bool = True,\n    ) -> None:\n        self._default = default\n        self._layout = layout\n        self._repaint = repaint\n        self._init = init\n        self._always_update = always_update\n        self._run_compute = compute\n\n    def __rich_repr__(self) -> rich.repr.Result:\n        yield self._default\n        yield \"layout\", self._layout\n        yield \"repaint\", self._repaint\n        yield \"init\", self._init\n        yield \"always_update\", self._always_update\n        yield \"compute\", self._run_compute\n\n    @classmethod\n    def var(\n        cls,\n        default: ReactiveType | Callable[[], ReactiveType],\n        always_update: bool = False,\n    ) -> Reactive:\n        \"\"\"A reactive variable that doesn't update or layout.\n\n        Args:\n            default: A default value or callable that returns a default.\n            always_update: Call watchers even when the new value equals the old value.\n\n        Returns:\n            A Reactive descriptor.\n        \"\"\"\n        return cls(default, layout=False, repaint=False, init=False)\n\n    def _initialize_reactive(self, obj: Reactable, name: str) -> None:\n        \"\"\"Initialized a reactive attribute on an object.\n\n        Args:\n            obj: An object with reactive attributes.\n            name: Name of attribute.\n        \"\"\"\n        _rich_traceback_omit = True\n        internal_name = f\"_reactive_{name}\"\n        if hasattr(obj, internal_name):\n            # Attribute already has a value\n            return\n\n        compute_method = getattr(obj, f\"compute_{name}\", None)\n        if compute_method is not None and self._init:\n            default = getattr(obj, f\"compute_{name}\")()\n        else:\n            default_or_callable = self._default\n            default = (\n                default_or_callable()\n                if callable(default_or_callable)\n                else default_or_callable\n            )\n        setattr(obj, internal_name, default)\n        if self._init:\n            self._check_watchers(obj, name, default)\n\n    @classmethod\n    def _initialize_object(cls, obj: Reactable) -> None:\n        \"\"\"Set defaults and call any watchers / computes for the first time.\n\n        Args:\n            obj: An object with Reactive descriptors\n        \"\"\"\n        _rich_traceback_omit = True\n        for name, reactive in obj._reactives.items():\n            reactive._initialize_reactive(obj, name)\n\n    @classmethod\n    def _reset_object(cls, obj: object) -> None:\n        \"\"\"Reset reactive structures on object (to avoid reference cycles).\n\n        Args:\n            obj: A reactive object.\n        \"\"\"\n        getattr(obj, \"__watchers\", {}).clear()\n        getattr(obj, \"__computes\", []).clear()\n\n    def __set_name__(self, owner: Type[MessageTarget], name: str) -> None:\n        # Check for compute method\n        if hasattr(owner, f\"compute_{name}\"):\n            # Compute methods are stored in a list called `__computes`\n            try:\n                computes = getattr(owner, \"__computes\")\n            except AttributeError:\n                computes = []\n                setattr(owner, \"__computes\", computes)\n            computes.append(name)\n\n        # The name of the attribute\n        self.name = name\n        # The internal name where the attribute's value is stored\n        self.internal_name = f\"_reactive_{name}\"\n        self.compute_name = f\"compute_{name}\"\n        default = self._default\n        setattr(owner, f\"_default_{name}\", default)\n\n    def __get__(self, obj: Reactable, obj_type: type[object]) -> ReactiveType:\n        internal_name = self.internal_name\n        if not hasattr(obj, internal_name):\n            self._initialize_reactive(obj, self.name)\n\n        if hasattr(obj, self.compute_name):\n            value: ReactiveType\n            old_value = getattr(obj, internal_name)\n            _rich_traceback_omit = True\n            value = getattr(obj, self.compute_name)()\n            setattr(obj, internal_name, value)\n            self._check_watchers(obj, self.name, old_value)\n            return value\n        else:\n            return getattr(obj, internal_name)\n\n    def __set__(self, obj: Reactable, value: ReactiveType) -> None:\n        _rich_traceback_omit = True\n\n        self._initialize_reactive(obj, self.name)\n\n        if hasattr(obj, self.compute_name):\n            raise AttributeError(\n                f\"Can't set {obj}.{self.name!r}; reactive attributes with a compute method are read-only\"\n            )\n\n        name = self.name\n        current_value = getattr(obj, name)\n        # Check for validate function\n        validate_function = getattr(obj, f\"validate_{name}\", None)\n        # Call validate\n        if callable(validate_function):\n            value = validate_function(value)\n        # If the value has changed, or this is the first time setting the value\n        if current_value != value or self._always_update:\n            # Store the internal value\n            setattr(obj, self.internal_name, value)\n\n            # Check all watchers\n            self._check_watchers(obj, name, current_value)\n\n            if self._run_compute:\n                self._compute(obj)\n\n            # Refresh according to descriptor flags\n            if self._layout or self._repaint:\n                obj.refresh(repaint=self._repaint, layout=self._layout)\n\n    @classmethod\n    def _check_watchers(cls, obj: Reactable, name: str, old_value: Any):\n        \"\"\"Check watchers, and call watch methods / computes\n\n        Args:\n            obj: The reactable object.\n            name: Attribute name.\n            old_value: The old (previous) value of the attribute.\n        \"\"\"\n        _rich_traceback_omit = True\n        # Get the current value.\n        internal_name = f\"_reactive_{name}\"\n        value = getattr(obj, internal_name)\n\n        async def await_watcher(awaitable: Awaitable) -> None:\n            \"\"\"Coroutine to await an awaitable returned from a watcher\"\"\"\n            _rich_traceback_omit = True\n            await awaitable\n            # Watcher may have changed the state, so run compute again\n            obj.post_message(events.Callback(callback=partial(Reactive._compute, obj)))\n\n        def invoke_watcher(\n            watch_function: Callable, old_value: object, value: object\n        ) -> None:\n            \"\"\"Invoke a watch function.\n\n            Args:\n                watch_function: A watch function, which may be sync or async.\n                old_value: The old value of the attribute.\n                value: The new value of the attribute.\n            \"\"\"\n            _rich_traceback_omit = True\n            param_count = count_parameters(watch_function)\n            if param_count == 2:\n                watch_result = watch_function(old_value, value)\n            elif param_count == 1:\n                watch_result = watch_function(value)\n            else:\n                watch_result = watch_function()\n            if isawaitable(watch_result):\n                # Result is awaitable, so we need to await it within an async context\n                obj.post_message(\n                    events.Callback(callback=partial(await_watcher, watch_result))\n                )\n\n        private_watch_function = getattr(obj, f\"_watch_{name}\", None)\n        if callable(private_watch_function):\n            invoke_watcher(private_watch_function, old_value, value)\n\n        public_watch_function = getattr(obj, f\"watch_{name}\", None)\n        if callable(public_watch_function):\n            invoke_watcher(public_watch_function, old_value, value)\n\n        # Process \"global\" watchers\n        watchers: list[tuple[Reactable, Callable]]\n        watchers = getattr(obj, \"__watchers\", {}).get(name, [])\n        # Remove any watchers for reactables that have since closed\n        if watchers:\n            watchers[:] = [\n                (reactable, callback)\n                for reactable, callback in watchers\n                if reactable.is_attached and not reactable._closing\n            ]\n            for reactable, callback in watchers:\n                with reactable.prevent(*obj._prevent_message_types_stack[-1]):\n                    invoke_watcher(callback, old_value, value)\n\n    @classmethod\n    def _compute(cls, obj: Reactable) -> None:\n        \"\"\"Invoke all computes.\n\n        Args:\n            obj: Reactable object.\n        \"\"\"\n        _rich_traceback_guard = True\n        for compute in obj._reactives.keys():\n            try:\n                compute_method = getattr(obj, f\"compute_{compute}\")\n            except AttributeError:\n                continue\n            current_value = getattr(\n                obj, f\"_reactive_{compute}\", getattr(obj, f\"_default_{compute}\", None)\n            )\n            value = compute_method()\n            setattr(obj, f\"_reactive_{compute}\", value)\n            if value != current_value:\n                cls._check_watchers(obj, compute, current_value)\n\n\nclass reactive(Reactive[ReactiveType]):\n    \"\"\"Create a reactive attribute.\n\n    Args:\n        default: A default value or callable that returns a default.\n        layout: Perform a layout on change.\n        repaint: Perform a repaint on change.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n    \"\"\"\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        *,\n        layout: bool = False,\n        repaint: bool = True,\n        init: bool = True,\n        always_update: bool = False,\n    ) -> None:\n        super().__init__(\n            default,\n            layout=layout,\n            repaint=repaint,\n            init=init,\n            always_update=always_update,\n        )\n\n\nclass var(Reactive[ReactiveType]):\n    \"\"\"Create a reactive attribute (with no auto-refresh).\n\n    Args:\n        default: A default value or callable that returns a default.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n    \"\"\"\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        init: bool = True,\n        always_update: bool = False,\n    ) -> None:\n        super().__init__(\n            default,\n            layout=False,\n            repaint=False,\n            init=init,\n            always_update=always_update,\n        )\n\n\ndef _watch(\n    node: DOMNode,\n    obj: Reactable,\n    attribute_name: str,\n    callback: WatchCallbackType,\n    *,\n    init: bool = True,\n) -> None:\n    \"\"\"Watch a reactive variable on an object.\n\n    Args:\n        obj: The parent object.\n        attribute_name: The attribute to watch.\n        callback: A callable to call when the attribute changes.\n        init: True to call watcher initialization.\n    \"\"\"\n    if not hasattr(obj, \"__watchers\"):\n        setattr(obj, \"__watchers\", {})\n    watchers: dict[str, list[tuple[Reactable, Callable]]] = getattr(obj, \"__watchers\")\n    watcher_list = watchers.setdefault(attribute_name, [])\n    if callback in watcher_list:\n        return\n    watcher_list.append((node, callback))\n    if init:\n        current_value = getattr(obj, attribute_name, None)\n        Reactive._check_watchers(obj, attribute_name, current_value)\n", "path": "src/textual/reactive.py"}], "after_files": [{"content": "\"\"\"\n\nThe `Reactive` class implements [reactivity](/guide/reactivity/).\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom functools import partial\nfrom inspect import isawaitable\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Awaitable,\n    Callable,\n    ClassVar,\n    Generic,\n    Type,\n    TypeVar,\n)\n\nimport rich.repr\n\nfrom . import events\nfrom ._callback import count_parameters\nfrom ._types import MessageTarget, WatchCallbackType\n\nif TYPE_CHECKING:\n    from .dom import DOMNode\n\n    Reactable = DOMNode\n\nReactiveType = TypeVar(\"ReactiveType\")\n\n\n@rich.repr.auto\nclass Reactive(Generic[ReactiveType]):\n    \"\"\"Reactive descriptor.\n\n    Args:\n        default: A default value or callable that returns a default.\n        layout: Perform a layout on change.\n        repaint: Perform a repaint on change.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n        compute: Run compute methods when attribute is changed.\n    \"\"\"\n\n    _reactives: ClassVar[dict[str, object]] = {}\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        *,\n        layout: bool = False,\n        repaint: bool = True,\n        init: bool = False,\n        always_update: bool = False,\n        compute: bool = True,\n    ) -> None:\n        self._default = default\n        self._layout = layout\n        self._repaint = repaint\n        self._init = init\n        self._always_update = always_update\n        self._run_compute = compute\n\n    def __rich_repr__(self) -> rich.repr.Result:\n        yield self._default\n        yield \"layout\", self._layout\n        yield \"repaint\", self._repaint\n        yield \"init\", self._init\n        yield \"always_update\", self._always_update\n        yield \"compute\", self._run_compute\n\n    def _initialize_reactive(self, obj: Reactable, name: str) -> None:\n        \"\"\"Initialized a reactive attribute on an object.\n\n        Args:\n            obj: An object with reactive attributes.\n            name: Name of attribute.\n        \"\"\"\n        _rich_traceback_omit = True\n        internal_name = f\"_reactive_{name}\"\n        if hasattr(obj, internal_name):\n            # Attribute already has a value\n            return\n\n        compute_method = getattr(obj, f\"compute_{name}\", None)\n        if compute_method is not None and self._init:\n            default = getattr(obj, f\"compute_{name}\")()\n        else:\n            default_or_callable = self._default\n            default = (\n                default_or_callable()\n                if callable(default_or_callable)\n                else default_or_callable\n            )\n        setattr(obj, internal_name, default)\n        if self._init:\n            self._check_watchers(obj, name, default)\n\n    @classmethod\n    def _initialize_object(cls, obj: Reactable) -> None:\n        \"\"\"Set defaults and call any watchers / computes for the first time.\n\n        Args:\n            obj: An object with Reactive descriptors\n        \"\"\"\n        _rich_traceback_omit = True\n        for name, reactive in obj._reactives.items():\n            reactive._initialize_reactive(obj, name)\n\n    @classmethod\n    def _reset_object(cls, obj: object) -> None:\n        \"\"\"Reset reactive structures on object (to avoid reference cycles).\n\n        Args:\n            obj: A reactive object.\n        \"\"\"\n        getattr(obj, \"__watchers\", {}).clear()\n        getattr(obj, \"__computes\", []).clear()\n\n    def __set_name__(self, owner: Type[MessageTarget], name: str) -> None:\n        # Check for compute method\n        if hasattr(owner, f\"compute_{name}\"):\n            # Compute methods are stored in a list called `__computes`\n            try:\n                computes = getattr(owner, \"__computes\")\n            except AttributeError:\n                computes = []\n                setattr(owner, \"__computes\", computes)\n            computes.append(name)\n\n        # The name of the attribute\n        self.name = name\n        # The internal name where the attribute's value is stored\n        self.internal_name = f\"_reactive_{name}\"\n        self.compute_name = f\"compute_{name}\"\n        default = self._default\n        setattr(owner, f\"_default_{name}\", default)\n\n    def __get__(self, obj: Reactable, obj_type: type[object]) -> ReactiveType:\n        internal_name = self.internal_name\n        if not hasattr(obj, internal_name):\n            self._initialize_reactive(obj, self.name)\n\n        if hasattr(obj, self.compute_name):\n            value: ReactiveType\n            old_value = getattr(obj, internal_name)\n            _rich_traceback_omit = True\n            value = getattr(obj, self.compute_name)()\n            setattr(obj, internal_name, value)\n            self._check_watchers(obj, self.name, old_value)\n            return value\n        else:\n            return getattr(obj, internal_name)\n\n    def __set__(self, obj: Reactable, value: ReactiveType) -> None:\n        _rich_traceback_omit = True\n\n        self._initialize_reactive(obj, self.name)\n\n        if hasattr(obj, self.compute_name):\n            raise AttributeError(\n                f\"Can't set {obj}.{self.name!r}; reactive attributes with a compute method are read-only\"\n            )\n\n        name = self.name\n        current_value = getattr(obj, name)\n        # Check for validate function\n        validate_function = getattr(obj, f\"validate_{name}\", None)\n        # Call validate\n        if callable(validate_function):\n            value = validate_function(value)\n        # If the value has changed, or this is the first time setting the value\n        if current_value != value or self._always_update:\n            # Store the internal value\n            setattr(obj, self.internal_name, value)\n\n            # Check all watchers\n            self._check_watchers(obj, name, current_value)\n\n            if self._run_compute:\n                self._compute(obj)\n\n            # Refresh according to descriptor flags\n            if self._layout or self._repaint:\n                obj.refresh(repaint=self._repaint, layout=self._layout)\n\n    @classmethod\n    def _check_watchers(cls, obj: Reactable, name: str, old_value: Any):\n        \"\"\"Check watchers, and call watch methods / computes\n\n        Args:\n            obj: The reactable object.\n            name: Attribute name.\n            old_value: The old (previous) value of the attribute.\n        \"\"\"\n        _rich_traceback_omit = True\n        # Get the current value.\n        internal_name = f\"_reactive_{name}\"\n        value = getattr(obj, internal_name)\n\n        async def await_watcher(awaitable: Awaitable) -> None:\n            \"\"\"Coroutine to await an awaitable returned from a watcher\"\"\"\n            _rich_traceback_omit = True\n            await awaitable\n            # Watcher may have changed the state, so run compute again\n            obj.post_message(events.Callback(callback=partial(Reactive._compute, obj)))\n\n        def invoke_watcher(\n            watch_function: Callable, old_value: object, value: object\n        ) -> None:\n            \"\"\"Invoke a watch function.\n\n            Args:\n                watch_function: A watch function, which may be sync or async.\n                old_value: The old value of the attribute.\n                value: The new value of the attribute.\n            \"\"\"\n            _rich_traceback_omit = True\n            param_count = count_parameters(watch_function)\n            if param_count == 2:\n                watch_result = watch_function(old_value, value)\n            elif param_count == 1:\n                watch_result = watch_function(value)\n            else:\n                watch_result = watch_function()\n            if isawaitable(watch_result):\n                # Result is awaitable, so we need to await it within an async context\n                obj.post_message(\n                    events.Callback(callback=partial(await_watcher, watch_result))\n                )\n\n        private_watch_function = getattr(obj, f\"_watch_{name}\", None)\n        if callable(private_watch_function):\n            invoke_watcher(private_watch_function, old_value, value)\n\n        public_watch_function = getattr(obj, f\"watch_{name}\", None)\n        if callable(public_watch_function):\n            invoke_watcher(public_watch_function, old_value, value)\n\n        # Process \"global\" watchers\n        watchers: list[tuple[Reactable, Callable]]\n        watchers = getattr(obj, \"__watchers\", {}).get(name, [])\n        # Remove any watchers for reactables that have since closed\n        if watchers:\n            watchers[:] = [\n                (reactable, callback)\n                for reactable, callback in watchers\n                if reactable.is_attached and not reactable._closing\n            ]\n            for reactable, callback in watchers:\n                with reactable.prevent(*obj._prevent_message_types_stack[-1]):\n                    invoke_watcher(callback, old_value, value)\n\n    @classmethod\n    def _compute(cls, obj: Reactable) -> None:\n        \"\"\"Invoke all computes.\n\n        Args:\n            obj: Reactable object.\n        \"\"\"\n        _rich_traceback_guard = True\n        for compute in obj._reactives.keys():\n            try:\n                compute_method = getattr(obj, f\"compute_{compute}\")\n            except AttributeError:\n                continue\n            current_value = getattr(\n                obj, f\"_reactive_{compute}\", getattr(obj, f\"_default_{compute}\", None)\n            )\n            value = compute_method()\n            setattr(obj, f\"_reactive_{compute}\", value)\n            if value != current_value:\n                cls._check_watchers(obj, compute, current_value)\n\n\nclass reactive(Reactive[ReactiveType]):\n    \"\"\"Create a reactive attribute.\n\n    Args:\n        default: A default value or callable that returns a default.\n        layout: Perform a layout on change.\n        repaint: Perform a repaint on change.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n    \"\"\"\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        *,\n        layout: bool = False,\n        repaint: bool = True,\n        init: bool = True,\n        always_update: bool = False,\n    ) -> None:\n        super().__init__(\n            default,\n            layout=layout,\n            repaint=repaint,\n            init=init,\n            always_update=always_update,\n        )\n\n\nclass var(Reactive[ReactiveType]):\n    \"\"\"Create a reactive attribute (with no auto-refresh).\n\n    Args:\n        default: A default value or callable that returns a default.\n        init: Call watchers on initialize (post mount).\n        always_update: Call watchers even when the new value equals the old value.\n    \"\"\"\n\n    def __init__(\n        self,\n        default: ReactiveType | Callable[[], ReactiveType],\n        init: bool = True,\n        always_update: bool = False,\n    ) -> None:\n        super().__init__(\n            default,\n            layout=False,\n            repaint=False,\n            init=init,\n            always_update=always_update,\n        )\n\n\ndef _watch(\n    node: DOMNode,\n    obj: Reactable,\n    attribute_name: str,\n    callback: WatchCallbackType,\n    *,\n    init: bool = True,\n) -> None:\n    \"\"\"Watch a reactive variable on an object.\n\n    Args:\n        obj: The parent object.\n        attribute_name: The attribute to watch.\n        callback: A callable to call when the attribute changes.\n        init: True to call watcher initialization.\n    \"\"\"\n    if not hasattr(obj, \"__watchers\"):\n        setattr(obj, \"__watchers\", {})\n    watchers: dict[str, list[tuple[Reactable, Callable]]] = getattr(obj, \"__watchers\")\n    watcher_list = watchers.setdefault(attribute_name, [])\n    if callback in watcher_list:\n        return\n    watcher_list.append((node, callback))\n    if init:\n        current_value = getattr(obj, attribute_name, None)\n        Reactive._check_watchers(obj, attribute_name, current_value)\n", "path": "src/textual/reactive.py"}]}