We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
linter complaint about using a worker as a timer callback
Have you checked closed issues? https://github.com/Textualize/textual/issues?q=is%3Aissue+is%3Aclosed

yes

Please give a brief but clear explanation of the issue. If you can, include a complete working example that demonstrates the bug. **Check it can run without modifications.**

```
from textual import work
from textual.app import App
class testy(App):
    @work
    async def worky(self) -> None:
        pass
    def compose(self):
        self.set_timer(1, self.worky)
```

```
$ pyright test.py
/home/guy/gits/ttuna/test.py
  /home/guy/gits/ttuna/test.py:8:27 - error: Argument of type "() -> Worker[None]" cannot be assigned to parameter "callback" of type "TimerCallback | None" in function "set_timer"
    Type "() -> Worker[None]" cannot be assigned to type "TimerCallback | None"
      Type "() -> Worker[None]" cannot be assigned to type "() -> Awaitable[None]"
        Function return type "Worker[None]" is incompatible with type "Awaitable[None]"
          "Worker[None]" is incompatible with protocol "Awaitable[None]"
            "__await__" is not present
      Type "() -> Worker[None]" cannot be assigned to type "() -> None"
        Function return type "Worker[None]" is incompatible with type "None"
          Type cannot be assigned to type "None"
    ... (reportGeneralTypeIssues)
1 error, 0 warnings, 0 informations 
```

The same error is reported for `set_interval` as well

--- 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/timer.py`
Content:
```
1 """
2 
3 Timer objects are created by [set_interval][textual.message_pump.MessagePump.set_interval] or
4     [set_timer][textual.message_pump.MessagePump.set_timer].
5 """
6 
7 from __future__ import annotations
8 
9 import weakref
10 from asyncio import CancelledError, Event, Task
11 from typing import Awaitable, Callable, Union
12 
13 from rich.repr import Result, rich_repr
14 
15 from . import _time, events
16 from ._asyncio import create_task
17 from ._callback import invoke
18 from ._context import active_app
19 from ._time import sleep
20 from ._types import MessageTarget
21 
22 TimerCallback = Union[Callable[[], Awaitable[None]], Callable[[], None]]
23 """Type of valid callbacks to be used with timers."""
24 
25 
26 class EventTargetGone(Exception):
27     pass
28 
29 
30 @rich_repr
31 class Timer:
32     """A class to send timer-based events.
33 
34     Args:
35         event_target: The object which will receive the timer events.
36         interval: The time between timer events, in seconds.
37         name: A name to assign the event (for debugging).
38         callback: A optional callback to invoke when the event is handled.
39         repeat: The number of times to repeat the timer, or None to repeat forever.
40         skip: Enable skipping of scheduled events that couldn't be sent in time.
41         pause: Start the timer paused.
42     """
43 
44     _timer_count: int = 1
45 
46     def __init__(
47         self,
48         event_target: MessageTarget,
49         interval: float,
50         *,
51         name: str | None = None,
52         callback: TimerCallback | None = None,
53         repeat: int | None = None,
54         skip: bool = True,
55         pause: bool = False,
56     ) -> None:
57         self._target_repr = repr(event_target)
58         self._target = weakref.ref(event_target)
59         self._interval = interval
60         self.name = f"Timer#{self._timer_count}" if name is None else name
61         self._timer_count += 1
62         self._callback = callback
63         self._repeat = repeat
64         self._skip = skip
65         self._active = Event()
66         self._task: Task | None = None
67         self._reset: bool = False
68         if not pause:
69             self._active.set()
70 
71     def __rich_repr__(self) -> Result:
72         yield self._interval
73         yield "name", self.name
74         yield "repeat", self._repeat, None
75 
76     @property
77     def target(self) -> MessageTarget:
78         target = self._target()
79         if target is None:
80             raise EventTargetGone()
81         return target
82 
83     def _start(self) -> None:
84         """Start the timer."""
85         self._task = create_task(self._run_timer(), name=self.name)
86 
87     def stop(self) -> None:
88         """Stop the timer."""
89         if self._task is not None:
90             self._active.set()
91             self._task.cancel()
92             self._task = None
93 
94     def pause(self) -> None:
95         """Pause the timer.
96 
97         A paused timer will not send events until it is resumed.
98         """
99         self._active.clear()
100 
101     def reset(self) -> None:
102         """Reset the timer, so it starts from the beginning."""
103         self._active.set()
104         self._reset = True
105 
106     def resume(self) -> None:
107         """Resume a paused timer."""
108         self._active.set()
109 
110     async def _run_timer(self) -> None:
111         """Run the timer task."""
112         try:
113             await self._run()
114         except CancelledError:
115             pass
116 
117     async def _run(self) -> None:
118         """Run the timer."""
119         count = 0
120         _repeat = self._repeat
121         _interval = self._interval
122         await self._active.wait()
123         start = _time.get_time()
124 
125         while _repeat is None or count <= _repeat:
126             next_timer = start + ((count + 1) * _interval)
127             now = _time.get_time()
128             if self._skip and next_timer < now:
129                 count += 1
130                 continue
131             now = _time.get_time()
132             wait_time = max(0, next_timer - now)
133             await sleep(wait_time)
134             count += 1
135             await self._active.wait()
136             if self._reset:
137                 start = _time.get_time()
138                 count = 0
139                 self._reset = False
140                 continue
141             try:
142                 await self._tick(next_timer=next_timer, count=count)
143             except EventTargetGone:
144                 break
145 
146     async def _tick(self, *, next_timer: float, count: int) -> None:
147         """Triggers the Timer's action: either call its callback, or sends an event to its target"""
148         if self._callback is not None:
149             try:
150                 await invoke(self._callback)
151             except CancelledError:
152                 # https://github.com/Textualize/textual/pull/2895
153                 # Re-raise CancelledErrors that would be caught by the following exception block in Python 3.7
154                 raise
155             except Exception as error:
156                 app = active_app.get()
157                 app._handle_exception(error)
158         else:
159             event = events.Timer(
160                 timer=self,
161                 time=next_timer,
162                 count=count,
163                 callback=self._callback,
164             )
165             self.target.post_message(event)
166
```

Path: `src/textual/_work_decorator.py`
Content:
```
1 """
2 
3 A decorator used to create [workers](/guide/workers).
4 """
5 
6 
7 from __future__ import annotations
8 
9 from functools import partial, wraps
10 from inspect import iscoroutinefunction
11 from typing import TYPE_CHECKING, Callable, Coroutine, TypeVar, Union, cast, overload
12 
13 from typing_extensions import ParamSpec, TypeAlias
14 
15 if TYPE_CHECKING:
16     from .worker import Worker
17 
18 
19 FactoryParamSpec = ParamSpec("FactoryParamSpec")
20 DecoratorParamSpec = ParamSpec("DecoratorParamSpec")
21 ReturnType = TypeVar("ReturnType")
22 
23 Decorator: TypeAlias = Callable[
24     [
25         Union[
26             Callable[DecoratorParamSpec, ReturnType],
27             Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]],
28         ]
29     ],
30     Callable[DecoratorParamSpec, "Worker[ReturnType]"],
31 ]
32 
33 
34 class WorkerDeclarationError(Exception):
35     """An error in the declaration of a worker method."""
36 
37 
38 @overload
39 def work(
40     method: Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]],
41     *,
42     name: str = "",
43     group: str = "default",
44     exit_on_error: bool = True,
45     exclusive: bool = False,
46     description: str | None = None,
47     thread: bool = False,
48 ) -> Callable[FactoryParamSpec, "Worker[ReturnType]"]:
49     ...
50 
51 
52 @overload
53 def work(
54     method: Callable[FactoryParamSpec, ReturnType],
55     *,
56     name: str = "",
57     group: str = "default",
58     exit_on_error: bool = True,
59     exclusive: bool = False,
60     description: str | None = None,
61     thread: bool = False,
62 ) -> Callable[FactoryParamSpec, "Worker[ReturnType]"]:
63     ...
64 
65 
66 @overload
67 def work(
68     *,
69     name: str = "",
70     group: str = "default",
71     exit_on_error: bool = True,
72     exclusive: bool = False,
73     description: str | None = None,
74     thread: bool = False,
75 ) -> Decorator[..., ReturnType]:
76     ...
77 
78 
79 def work(
80     method: Callable[FactoryParamSpec, ReturnType]
81     | Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]]
82     | None = None,
83     *,
84     name: str = "",
85     group: str = "default",
86     exit_on_error: bool = True,
87     exclusive: bool = False,
88     description: str | None = None,
89     thread: bool = False,
90 ) -> Callable[FactoryParamSpec, Worker[ReturnType]] | Decorator:
91     """A decorator used to create [workers](/guide/workers).
92 
93     Args:
94         method: A function or coroutine.
95         name: A short string to identify the worker (in logs and debugging).
96         group: A short string to identify a group of workers.
97         exit_on_error: Exit the app if the worker raises an error. Set to `False` to suppress exceptions.
98         exclusive: Cancel all workers in the same group.
99         description: Readable description of the worker for debugging purposes.
100             By default, it uses a string representation of the decorated method
101             and its arguments.
102         thread: Mark the method as a thread worker.
103     """
104 
105     def decorator(
106         method: (
107             Callable[DecoratorParamSpec, ReturnType]
108             | Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]]
109         )
110     ) -> Callable[DecoratorParamSpec, Worker[ReturnType]]:
111         """The decorator."""
112 
113         # Methods that aren't async *must* be marked as being a thread
114         # worker.
115         if not iscoroutinefunction(method) and not thread:
116             raise WorkerDeclarationError(
117                 "Can not create a worker from a non-async function unless `thread=True` is set on the work decorator."
118             )
119 
120         @wraps(method)
121         def decorated(
122             *args: DecoratorParamSpec.args, **kwargs: DecoratorParamSpec.kwargs
123         ) -> Worker[ReturnType]:
124             """The replaced callable."""
125             from .dom import DOMNode
126 
127             self = args[0]
128             assert isinstance(self, DOMNode)
129 
130             if description is not None:
131                 debug_description = description
132             else:
133                 try:
134                     positional_arguments = ", ".join(repr(arg) for arg in args[1:])
135                     keyword_arguments = ", ".join(
136                         f"{name}={value!r}" for name, value in kwargs.items()
137                     )
138                     tokens = [positional_arguments, keyword_arguments]
139                     debug_description = f"{method.__name__}({', '.join(token for token in tokens if token)})"
140                 except Exception:
141                     debug_description = "<worker>"
142             worker = cast(
143                 "Worker[ReturnType]",
144                 self.run_worker(
145                     partial(method, *args, **kwargs),
146                     name=name or method.__name__,
147                     group=group,
148                     description=debug_description,
149                     exclusive=exclusive,
150                     exit_on_error=exit_on_error,
151                     thread=thread,
152                 ),
153             )
154             return worker
155 
156         return decorated
157 
158     if method is None:
159         return decorator
160     else:
161         return decorator(method)
162
```
--- 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/_work_decorator.py b/src/textual/_work_decorator.py
--- a/src/textual/_work_decorator.py
+++ b/src/textual/_work_decorator.py
@@ -1,5 +1,4 @@
 """
-
 A decorator used to create [workers](/guide/workers).
 """
 
diff --git a/src/textual/timer.py b/src/textual/timer.py
--- a/src/textual/timer.py
+++ b/src/textual/timer.py
@@ -8,7 +8,7 @@
 
 import weakref
 from asyncio import CancelledError, Event, Task
-from typing import Awaitable, Callable, Union
+from typing import Any, Awaitable, Callable, Union
 
 from rich.repr import Result, rich_repr
 
@@ -19,7 +19,7 @@
 from ._time import sleep
 from ._types import MessageTarget
 
-TimerCallback = Union[Callable[[], Awaitable[None]], Callable[[], None]]
+TimerCallback = Union[Callable[[], Awaitable[Any]], Callable[[], Any]]
 """Type of valid callbacks to be used with timers."""

{"golden_diff": "diff --git a/src/textual/_work_decorator.py b/src/textual/_work_decorator.py\n--- a/src/textual/_work_decorator.py\n+++ b/src/textual/_work_decorator.py\n@@ -1,5 +1,4 @@\n \"\"\"\n-\n A decorator used to create [workers](/guide/workers).\n \"\"\"\n \ndiff --git a/src/textual/timer.py b/src/textual/timer.py\n--- a/src/textual/timer.py\n+++ b/src/textual/timer.py\n@@ -8,7 +8,7 @@\n \n import weakref\n from asyncio import CancelledError, Event, Task\n-from typing import Awaitable, Callable, Union\n+from typing import Any, Awaitable, Callable, Union\n \n from rich.repr import Result, rich_repr\n \n@@ -19,7 +19,7 @@\n from ._time import sleep\n from ._types import MessageTarget\n \n-TimerCallback = Union[Callable[[], Awaitable[None]], Callable[[], None]]\n+TimerCallback = Union[Callable[[], Awaitable[Any]], Callable[[], Any]]\n \"\"\"Type of valid callbacks to be used with timers.\"\"\"\n", "issue": "linter complaint about using a worker as a timer callback\nHave you checked closed issues? https://github.com/Textualize/textual/issues?q=is%3Aissue+is%3Aclosed\r\n\r\nyes\r\n\r\nPlease give a brief but clear explanation of the issue. If you can, include a complete working example that demonstrates the bug. **Check it can run without modifications.**\r\n\r\n```\r\nfrom textual import work\r\nfrom textual.app import App\r\nclass testy(App):\r\n    @work\r\n    async def worky(self) -> None:\r\n        pass\r\n    def compose(self):\r\n        self.set_timer(1, self.worky)\r\n```\r\n\r\n```\r\n$ pyright test.py\r\n/home/guy/gits/ttuna/test.py\r\n  /home/guy/gits/ttuna/test.py:8:27 - error: Argument of type \"() -> Worker[None]\" cannot be assigned to parameter \"callback\" of type \"TimerCallback | None\" in function \"set_timer\"\r\n  \u00a0\u00a0Type \"() -> Worker[None]\" cannot be assigned to type \"TimerCallback | None\"\r\n  \u00a0\u00a0\u00a0\u00a0Type \"() -> Worker[None]\" cannot be assigned to type \"() -> Awaitable[None]\"\r\n  \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0Function return type \"Worker[None]\" is incompatible with type \"Awaitable[None]\"\r\n  \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\"Worker[None]\" is incompatible with protocol \"Awaitable[None]\"\r\n  \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\"__await__\" is not present\r\n  \u00a0\u00a0\u00a0\u00a0Type \"() -> Worker[None]\" cannot be assigned to type \"() -> None\"\r\n  \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0Function return type \"Worker[None]\" is incompatible with type \"None\"\r\n  \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0Type cannot be assigned to type \"None\"\r\n    ... (reportGeneralTypeIssues)\r\n1 error, 0 warnings, 0 informations \r\n```\r\n\r\nThe same error is reported for `set_interval` as well\n", "before_files": [{"content": "\"\"\"\n\nTimer objects are created by [set_interval][textual.message_pump.MessagePump.set_interval] or\n    [set_timer][textual.message_pump.MessagePump.set_timer].\n\"\"\"\n\nfrom __future__ import annotations\n\nimport weakref\nfrom asyncio import CancelledError, Event, Task\nfrom typing import Awaitable, Callable, Union\n\nfrom rich.repr import Result, rich_repr\n\nfrom . import _time, events\nfrom ._asyncio import create_task\nfrom ._callback import invoke\nfrom ._context import active_app\nfrom ._time import sleep\nfrom ._types import MessageTarget\n\nTimerCallback = Union[Callable[[], Awaitable[None]], Callable[[], None]]\n\"\"\"Type of valid callbacks to be used with timers.\"\"\"\n\n\nclass EventTargetGone(Exception):\n    pass\n\n\n@rich_repr\nclass Timer:\n    \"\"\"A class to send timer-based events.\n\n    Args:\n        event_target: The object which will receive the timer events.\n        interval: The time between timer events, in seconds.\n        name: A name to assign the event (for debugging).\n        callback: A optional callback to invoke when the event is handled.\n        repeat: The number of times to repeat the timer, or None to repeat forever.\n        skip: Enable skipping of scheduled events that couldn't be sent in time.\n        pause: Start the timer paused.\n    \"\"\"\n\n    _timer_count: int = 1\n\n    def __init__(\n        self,\n        event_target: MessageTarget,\n        interval: float,\n        *,\n        name: str | None = None,\n        callback: TimerCallback | None = None,\n        repeat: int | None = None,\n        skip: bool = True,\n        pause: bool = False,\n    ) -> None:\n        self._target_repr = repr(event_target)\n        self._target = weakref.ref(event_target)\n        self._interval = interval\n        self.name = f\"Timer#{self._timer_count}\" if name is None else name\n        self._timer_count += 1\n        self._callback = callback\n        self._repeat = repeat\n        self._skip = skip\n        self._active = Event()\n        self._task: Task | None = None\n        self._reset: bool = False\n        if not pause:\n            self._active.set()\n\n    def __rich_repr__(self) -> Result:\n        yield self._interval\n        yield \"name\", self.name\n        yield \"repeat\", self._repeat, None\n\n    @property\n    def target(self) -> MessageTarget:\n        target = self._target()\n        if target is None:\n            raise EventTargetGone()\n        return target\n\n    def _start(self) -> None:\n        \"\"\"Start the timer.\"\"\"\n        self._task = create_task(self._run_timer(), name=self.name)\n\n    def stop(self) -> None:\n        \"\"\"Stop the timer.\"\"\"\n        if self._task is not None:\n            self._active.set()\n            self._task.cancel()\n            self._task = None\n\n    def pause(self) -> None:\n        \"\"\"Pause the timer.\n\n        A paused timer will not send events until it is resumed.\n        \"\"\"\n        self._active.clear()\n\n    def reset(self) -> None:\n        \"\"\"Reset the timer, so it starts from the beginning.\"\"\"\n        self._active.set()\n        self._reset = True\n\n    def resume(self) -> None:\n        \"\"\"Resume a paused timer.\"\"\"\n        self._active.set()\n\n    async def _run_timer(self) -> None:\n        \"\"\"Run the timer task.\"\"\"\n        try:\n            await self._run()\n        except CancelledError:\n            pass\n\n    async def _run(self) -> None:\n        \"\"\"Run the timer.\"\"\"\n        count = 0\n        _repeat = self._repeat\n        _interval = self._interval\n        await self._active.wait()\n        start = _time.get_time()\n\n        while _repeat is None or count <= _repeat:\n            next_timer = start + ((count + 1) * _interval)\n            now = _time.get_time()\n            if self._skip and next_timer < now:\n                count += 1\n                continue\n            now = _time.get_time()\n            wait_time = max(0, next_timer - now)\n            await sleep(wait_time)\n            count += 1\n            await self._active.wait()\n            if self._reset:\n                start = _time.get_time()\n                count = 0\n                self._reset = False\n                continue\n            try:\n                await self._tick(next_timer=next_timer, count=count)\n            except EventTargetGone:\n                break\n\n    async def _tick(self, *, next_timer: float, count: int) -> None:\n        \"\"\"Triggers the Timer's action: either call its callback, or sends an event to its target\"\"\"\n        if self._callback is not None:\n            try:\n                await invoke(self._callback)\n            except CancelledError:\n                # https://github.com/Textualize/textual/pull/2895\n                # Re-raise CancelledErrors that would be caught by the following exception block in Python 3.7\n                raise\n            except Exception as error:\n                app = active_app.get()\n                app._handle_exception(error)\n        else:\n            event = events.Timer(\n                timer=self,\n                time=next_timer,\n                count=count,\n                callback=self._callback,\n            )\n            self.target.post_message(event)\n", "path": "src/textual/timer.py"}, {"content": "\"\"\"\n\nA decorator used to create [workers](/guide/workers).\n\"\"\"\n\n\nfrom __future__ import annotations\n\nfrom functools import partial, wraps\nfrom inspect import iscoroutinefunction\nfrom typing import TYPE_CHECKING, Callable, Coroutine, TypeVar, Union, cast, overload\n\nfrom typing_extensions import ParamSpec, TypeAlias\n\nif TYPE_CHECKING:\n    from .worker import Worker\n\n\nFactoryParamSpec = ParamSpec(\"FactoryParamSpec\")\nDecoratorParamSpec = ParamSpec(\"DecoratorParamSpec\")\nReturnType = TypeVar(\"ReturnType\")\n\nDecorator: TypeAlias = Callable[\n    [\n        Union[\n            Callable[DecoratorParamSpec, ReturnType],\n            Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]],\n        ]\n    ],\n    Callable[DecoratorParamSpec, \"Worker[ReturnType]\"],\n]\n\n\nclass WorkerDeclarationError(Exception):\n    \"\"\"An error in the declaration of a worker method.\"\"\"\n\n\n@overload\ndef work(\n    method: Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]],\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, \"Worker[ReturnType]\"]:\n    ...\n\n\n@overload\ndef work(\n    method: Callable[FactoryParamSpec, ReturnType],\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, \"Worker[ReturnType]\"]:\n    ...\n\n\n@overload\ndef work(\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Decorator[..., ReturnType]:\n    ...\n\n\ndef work(\n    method: Callable[FactoryParamSpec, ReturnType]\n    | Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]]\n    | None = None,\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, Worker[ReturnType]] | Decorator:\n    \"\"\"A decorator used to create [workers](/guide/workers).\n\n    Args:\n        method: A function or coroutine.\n        name: A short string to identify the worker (in logs and debugging).\n        group: A short string to identify a group of workers.\n        exit_on_error: Exit the app if the worker raises an error. Set to `False` to suppress exceptions.\n        exclusive: Cancel all workers in the same group.\n        description: Readable description of the worker for debugging purposes.\n            By default, it uses a string representation of the decorated method\n            and its arguments.\n        thread: Mark the method as a thread worker.\n    \"\"\"\n\n    def decorator(\n        method: (\n            Callable[DecoratorParamSpec, ReturnType]\n            | Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]]\n        )\n    ) -> Callable[DecoratorParamSpec, Worker[ReturnType]]:\n        \"\"\"The decorator.\"\"\"\n\n        # Methods that aren't async *must* be marked as being a thread\n        # worker.\n        if not iscoroutinefunction(method) and not thread:\n            raise WorkerDeclarationError(\n                \"Can not create a worker from a non-async function unless `thread=True` is set on the work decorator.\"\n            )\n\n        @wraps(method)\n        def decorated(\n            *args: DecoratorParamSpec.args, **kwargs: DecoratorParamSpec.kwargs\n        ) -> Worker[ReturnType]:\n            \"\"\"The replaced callable.\"\"\"\n            from .dom import DOMNode\n\n            self = args[0]\n            assert isinstance(self, DOMNode)\n\n            if description is not None:\n                debug_description = description\n            else:\n                try:\n                    positional_arguments = \", \".join(repr(arg) for arg in args[1:])\n                    keyword_arguments = \", \".join(\n                        f\"{name}={value!r}\" for name, value in kwargs.items()\n                    )\n                    tokens = [positional_arguments, keyword_arguments]\n                    debug_description = f\"{method.__name__}({', '.join(token for token in tokens if token)})\"\n                except Exception:\n                    debug_description = \"<worker>\"\n            worker = cast(\n                \"Worker[ReturnType]\",\n                self.run_worker(\n                    partial(method, *args, **kwargs),\n                    name=name or method.__name__,\n                    group=group,\n                    description=debug_description,\n                    exclusive=exclusive,\n                    exit_on_error=exit_on_error,\n                    thread=thread,\n                ),\n            )\n            return worker\n\n        return decorated\n\n    if method is None:\n        return decorator\n    else:\n        return decorator(method)\n", "path": "src/textual/_work_decorator.py"}], "after_files": [{"content": "\"\"\"\n\nTimer objects are created by [set_interval][textual.message_pump.MessagePump.set_interval] or\n    [set_timer][textual.message_pump.MessagePump.set_timer].\n\"\"\"\n\nfrom __future__ import annotations\n\nimport weakref\nfrom asyncio import CancelledError, Event, Task\nfrom typing import Any, Awaitable, Callable, Union\n\nfrom rich.repr import Result, rich_repr\n\nfrom . import _time, events\nfrom ._asyncio import create_task\nfrom ._callback import invoke\nfrom ._context import active_app\nfrom ._time import sleep\nfrom ._types import MessageTarget\n\nTimerCallback = Union[Callable[[], Awaitable[Any]], Callable[[], Any]]\n\"\"\"Type of valid callbacks to be used with timers.\"\"\"\n\n\nclass EventTargetGone(Exception):\n    pass\n\n\n@rich_repr\nclass Timer:\n    \"\"\"A class to send timer-based events.\n\n    Args:\n        event_target: The object which will receive the timer events.\n        interval: The time between timer events, in seconds.\n        name: A name to assign the event (for debugging).\n        callback: A optional callback to invoke when the event is handled.\n        repeat: The number of times to repeat the timer, or None to repeat forever.\n        skip: Enable skipping of scheduled events that couldn't be sent in time.\n        pause: Start the timer paused.\n    \"\"\"\n\n    _timer_count: int = 1\n\n    def __init__(\n        self,\n        event_target: MessageTarget,\n        interval: float,\n        *,\n        name: str | None = None,\n        callback: TimerCallback | None = None,\n        repeat: int | None = None,\n        skip: bool = True,\n        pause: bool = False,\n    ) -> None:\n        self._target_repr = repr(event_target)\n        self._target = weakref.ref(event_target)\n        self._interval = interval\n        self.name = f\"Timer#{self._timer_count}\" if name is None else name\n        self._timer_count += 1\n        self._callback = callback\n        self._repeat = repeat\n        self._skip = skip\n        self._active = Event()\n        self._task: Task | None = None\n        self._reset: bool = False\n        if not pause:\n            self._active.set()\n\n    def __rich_repr__(self) -> Result:\n        yield self._interval\n        yield \"name\", self.name\n        yield \"repeat\", self._repeat, None\n\n    @property\n    def target(self) -> MessageTarget:\n        target = self._target()\n        if target is None:\n            raise EventTargetGone()\n        return target\n\n    def _start(self) -> None:\n        \"\"\"Start the timer.\"\"\"\n        self._task = create_task(self._run_timer(), name=self.name)\n\n    def stop(self) -> None:\n        \"\"\"Stop the timer.\"\"\"\n        if self._task is not None:\n            self._active.set()\n            self._task.cancel()\n            self._task = None\n\n    def pause(self) -> None:\n        \"\"\"Pause the timer.\n\n        A paused timer will not send events until it is resumed.\n        \"\"\"\n        self._active.clear()\n\n    def reset(self) -> None:\n        \"\"\"Reset the timer, so it starts from the beginning.\"\"\"\n        self._active.set()\n        self._reset = True\n\n    def resume(self) -> None:\n        \"\"\"Resume a paused timer.\"\"\"\n        self._active.set()\n\n    async def _run_timer(self) -> None:\n        \"\"\"Run the timer task.\"\"\"\n        try:\n            await self._run()\n        except CancelledError:\n            pass\n\n    async def _run(self) -> None:\n        \"\"\"Run the timer.\"\"\"\n        count = 0\n        _repeat = self._repeat\n        _interval = self._interval\n        await self._active.wait()\n        start = _time.get_time()\n\n        while _repeat is None or count <= _repeat:\n            next_timer = start + ((count + 1) * _interval)\n            now = _time.get_time()\n            if self._skip and next_timer < now:\n                count += 1\n                continue\n            now = _time.get_time()\n            wait_time = max(0, next_timer - now)\n            await sleep(wait_time)\n            count += 1\n            await self._active.wait()\n            if self._reset:\n                start = _time.get_time()\n                count = 0\n                self._reset = False\n                continue\n            try:\n                await self._tick(next_timer=next_timer, count=count)\n            except EventTargetGone:\n                break\n\n    async def _tick(self, *, next_timer: float, count: int) -> None:\n        \"\"\"Triggers the Timer's action: either call its callback, or sends an event to its target\"\"\"\n        if self._callback is not None:\n            try:\n                await invoke(self._callback)\n            except CancelledError:\n                # https://github.com/Textualize/textual/pull/2895\n                # Re-raise CancelledErrors that would be caught by the following exception block in Python 3.7\n                raise\n            except Exception as error:\n                app = active_app.get()\n                app._handle_exception(error)\n        else:\n            event = events.Timer(\n                timer=self,\n                time=next_timer,\n                count=count,\n                callback=self._callback,\n            )\n            self.target.post_message(event)\n", "path": "src/textual/timer.py"}, {"content": "\"\"\"\nA decorator used to create [workers](/guide/workers).\n\"\"\"\n\n\nfrom __future__ import annotations\n\nfrom functools import partial, wraps\nfrom inspect import iscoroutinefunction\nfrom typing import TYPE_CHECKING, Callable, Coroutine, TypeVar, Union, cast, overload\n\nfrom typing_extensions import ParamSpec, TypeAlias\n\nif TYPE_CHECKING:\n    from .worker import Worker\n\n\nFactoryParamSpec = ParamSpec(\"FactoryParamSpec\")\nDecoratorParamSpec = ParamSpec(\"DecoratorParamSpec\")\nReturnType = TypeVar(\"ReturnType\")\n\nDecorator: TypeAlias = Callable[\n    [\n        Union[\n            Callable[DecoratorParamSpec, ReturnType],\n            Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]],\n        ]\n    ],\n    Callable[DecoratorParamSpec, \"Worker[ReturnType]\"],\n]\n\n\nclass WorkerDeclarationError(Exception):\n    \"\"\"An error in the declaration of a worker method.\"\"\"\n\n\n@overload\ndef work(\n    method: Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]],\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, \"Worker[ReturnType]\"]:\n    ...\n\n\n@overload\ndef work(\n    method: Callable[FactoryParamSpec, ReturnType],\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, \"Worker[ReturnType]\"]:\n    ...\n\n\n@overload\ndef work(\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Decorator[..., ReturnType]:\n    ...\n\n\ndef work(\n    method: Callable[FactoryParamSpec, ReturnType]\n    | Callable[FactoryParamSpec, Coroutine[None, None, ReturnType]]\n    | None = None,\n    *,\n    name: str = \"\",\n    group: str = \"default\",\n    exit_on_error: bool = True,\n    exclusive: bool = False,\n    description: str | None = None,\n    thread: bool = False,\n) -> Callable[FactoryParamSpec, Worker[ReturnType]] | Decorator:\n    \"\"\"A decorator used to create [workers](/guide/workers).\n\n    Args:\n        method: A function or coroutine.\n        name: A short string to identify the worker (in logs and debugging).\n        group: A short string to identify a group of workers.\n        exit_on_error: Exit the app if the worker raises an error. Set to `False` to suppress exceptions.\n        exclusive: Cancel all workers in the same group.\n        description: Readable description of the worker for debugging purposes.\n            By default, it uses a string representation of the decorated method\n            and its arguments.\n        thread: Mark the method as a thread worker.\n    \"\"\"\n\n    def decorator(\n        method: (\n            Callable[DecoratorParamSpec, ReturnType]\n            | Callable[DecoratorParamSpec, Coroutine[None, None, ReturnType]]\n        )\n    ) -> Callable[DecoratorParamSpec, Worker[ReturnType]]:\n        \"\"\"The decorator.\"\"\"\n\n        # Methods that aren't async *must* be marked as being a thread\n        # worker.\n        if not iscoroutinefunction(method) and not thread:\n            raise WorkerDeclarationError(\n                \"Can not create a worker from a non-async function unless `thread=True` is set on the work decorator.\"\n            )\n\n        @wraps(method)\n        def decorated(\n            *args: DecoratorParamSpec.args, **kwargs: DecoratorParamSpec.kwargs\n        ) -> Worker[ReturnType]:\n            \"\"\"The replaced callable.\"\"\"\n            from .dom import DOMNode\n\n            self = args[0]\n            assert isinstance(self, DOMNode)\n\n            if description is not None:\n                debug_description = description\n            else:\n                try:\n                    positional_arguments = \", \".join(repr(arg) for arg in args[1:])\n                    keyword_arguments = \", \".join(\n                        f\"{name}={value!r}\" for name, value in kwargs.items()\n                    )\n                    tokens = [positional_arguments, keyword_arguments]\n                    debug_description = f\"{method.__name__}({', '.join(token for token in tokens if token)})\"\n                except Exception:\n                    debug_description = \"<worker>\"\n            worker = cast(\n                \"Worker[ReturnType]\",\n                self.run_worker(\n                    partial(method, *args, **kwargs),\n                    name=name or method.__name__,\n                    group=group,\n                    description=debug_description,\n                    exclusive=exclusive,\n                    exit_on_error=exit_on_error,\n                    thread=thread,\n                ),\n            )\n            return worker\n\n        return decorated\n\n    if method is None:\n        return decorator\n    else:\n        return decorator(method)\n", "path": "src/textual/_work_decorator.py"}]}