Ir para o conteúdo

tempestweb.devserver

O servidor de desenvolvimento por trás do tempestweb dev: observa arquivos e recarrega, sem saber qual modo está servindo. Interno — a referência está aqui para quem depura o fluxo de dev.

Guia com exemplos: Usando a CLI.

tempestweb.devserver

tempestweb dev server — transport-agnostic file watch + reload.

See docs/plan.md §5. This package owns the two halves of the dev loop:

  • :class:ReloadSignal — a publish/subscribe hub that decouples "something changed" from "how the reload reaches the app". A transport subscribes; the watcher (or the interactive cockpit) triggers.
  • :class:FileWatcher — observes a project directory and triggers the signal on every reload-worthy change.

Neither half names a transport. Mode A wires the signal to a browser reload; Mode B wires it to a session restart. The watcher and the signal stay identical.

ReloadEvent dataclass

A single reload notification.

Attributes:

Name Type Description
kind ReloadKind

Whether to restart (clean state) or reload (preserve state).
paths tuple[str, ...]

The project-relative paths whose change triggered the reload. Empty when the reload was triggered manually (e.g. the R key).
generation int

A monotonically increasing counter, starting at 1, that lets a consumer detect missed reloads after a slow tick.
Source code in tempestweb/devserver/reload.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True, slots=True)
class ReloadEvent:
    """A single reload notification.

    Attributes:
        kind: Whether to restart (clean state) or reload (preserve state).
        paths: The project-relative paths whose change triggered the reload.
            Empty when the reload was triggered manually (e.g. the ``R`` key).
        generation: A monotonically increasing counter, starting at 1, that lets
            a consumer detect missed reloads after a slow tick.
    """

    kind: ReloadKind = ReloadKind.RESTART
    paths: tuple[str, ...] = ()
    generation: int = 0

ReloadKind

Bases: StrEnum

The kind of reload a change should trigger.

Attributes:

Name Type Description
RESTART

Re-run the app from scratch with clean state (the v1 default — "hot restart"). See docs/plan.md §5.1.
RELOAD

Re-run preserving state ("hot reload"). Reserved for post-v1; the watcher never emits this yet but the type exists so transports can branch on it ahead of time.
Source code in tempestweb/devserver/reload.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
class ReloadKind(StrEnum):
    """The kind of reload a change should trigger.

    Attributes:
        RESTART: Re-run the app from scratch with clean state (the v1 default —
            "hot restart"). See ``docs/plan.md`` §5.1.
        RELOAD: Re-run preserving state ("hot reload"). Reserved for post-v1;
            the watcher never emits this yet but the type exists so transports
            can branch on it ahead of time.
    """

    RESTART = "restart"
    RELOAD = "reload"

ReloadSignal dataclass

A transport-agnostic publish/subscribe hub for reload events.

The watcher (or the interactive cockpit) is the producer; a transport is the consumer. Neither side imports the other — they meet at this object.

Example

signal = ReloadSignal() seen: list[ReloadEvent] = [] unsubscribe = signal.subscribe(seen.append) event = signal.trigger(paths=["app.py"]) seen[0] is event True event.generation 1 unsubscribe()

Source code in tempestweb/devserver/reload.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@dataclass(slots=True)
class ReloadSignal:
    """A transport-agnostic publish/subscribe hub for reload events.

    The watcher (or the interactive cockpit) is the producer; a transport is the
    consumer. Neither side imports the other — they meet at this object.

    Example:
        >>> signal = ReloadSignal()
        >>> seen: list[ReloadEvent] = []
        >>> unsubscribe = signal.subscribe(seen.append)
        >>> event = signal.trigger(paths=["app.py"])
        >>> seen[0] is event
        True
        >>> event.generation
        1
        >>> unsubscribe()
    """

    _generation: int = 0
    _callbacks: list[ReloadCallback] = field(default_factory=list)
    _waiters: list[asyncio.Future[ReloadEvent]] = field(default_factory=list)

    @property
    def generation(self) -> int:
        """Return the number of reloads emitted so far.

        Returns:
            The current generation counter (0 before the first reload).
        """
        return self._generation

    def subscribe(self, callback: ReloadCallback) -> Callable[[], None]:
        """Register a synchronous callback invoked on every reload.

        Args:
            callback: A function called with each :class:`ReloadEvent`.

        Returns:
            A zero-argument function that unregisters the callback when called.
        """
        self._callbacks.append(callback)

        def unsubscribe() -> None:
            """Remove the callback if it is still registered."""
            if callback in self._callbacks:
                self._callbacks.remove(callback)

        return unsubscribe

    def trigger(
        self,
        *,
        kind: ReloadKind = ReloadKind.RESTART,
        paths: list[str] | tuple[str, ...] = (),
    ) -> ReloadEvent:
        """Emit a reload event to every subscriber and waiter.

        Increments the generation counter, builds a :class:`ReloadEvent`, invokes
        every registered callback synchronously, and resolves any pending
        :meth:`wait` futures.

        Args:
            kind: The reload kind. Defaults to :attr:`ReloadKind.RESTART`.
            paths: The paths whose change caused the reload. Defaults to empty
                (a manual reload).

        Returns:
            The emitted :class:`ReloadEvent`.
        """
        self._generation += 1
        event = ReloadEvent(
            kind=kind,
            paths=tuple(paths),
            generation=self._generation,
        )
        for callback in list(self._callbacks):
            callback(event)
        waiters = self._waiters
        self._waiters = []
        for waiter in waiters:
            if not waiter.done():
                waiter.set_result(event)
        return event

    async def wait(self) -> ReloadEvent:
        """Await the next reload event.

        Returns:
            The next :class:`ReloadEvent` emitted by :meth:`trigger`.
        """
        loop = asyncio.get_running_loop()
        future: asyncio.Future[ReloadEvent] = loop.create_future()
        self._waiters.append(future)
        return await future

generation property 

generation: int

Return the number of reloads emitted so far.

Returns:

Type Description
int

The current generation counter (0 before the first reload).

subscribe 

subscribe(callback: ReloadCallback) -> Callable[[], None]

Register a synchronous callback invoked on every reload.

Parameters:

Name Type Description Default
callback ReloadCallback

A function called with each :class:ReloadEvent.

 required

Returns:

Type Description
Callable[[], None]

A zero-argument function that unregisters the callback when called.
Source code in tempestweb/devserver/reload.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def subscribe(self, callback: ReloadCallback) -> Callable[[], None]:
    """Register a synchronous callback invoked on every reload.

    Args:
        callback: A function called with each :class:`ReloadEvent`.

    Returns:
        A zero-argument function that unregisters the callback when called.
    """
    self._callbacks.append(callback)

    def unsubscribe() -> None:
        """Remove the callback if it is still registered."""
        if callback in self._callbacks:
            self._callbacks.remove(callback)

    return unsubscribe

trigger 

trigger(*, kind: ReloadKind = ReloadKind.RESTART, paths: list[str] | tuple[str, ...] = ()) -> ReloadEvent

Emit a reload event to every subscriber and waiter.

Increments the generation counter, builds a :class:ReloadEvent, invokes every registered callback synchronously, and resolves any pending :meth:wait futures.

Parameters:

Name Type Description Default
kind ReloadKind

The reload kind. Defaults to :attr:ReloadKind.RESTART.

 RESTART
paths list[str] | tuple[str, ...]

The paths whose change caused the reload. Defaults to empty (a manual reload).

 ()

Returns:

Type Description
ReloadEvent

The emitted :class:ReloadEvent.
Source code in tempestweb/devserver/reload.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def trigger(
    self,
    *,
    kind: ReloadKind = ReloadKind.RESTART,
    paths: list[str] | tuple[str, ...] = (),
) -> ReloadEvent:
    """Emit a reload event to every subscriber and waiter.

    Increments the generation counter, builds a :class:`ReloadEvent`, invokes
    every registered callback synchronously, and resolves any pending
    :meth:`wait` futures.

    Args:
        kind: The reload kind. Defaults to :attr:`ReloadKind.RESTART`.
        paths: The paths whose change caused the reload. Defaults to empty
            (a manual reload).

    Returns:
        The emitted :class:`ReloadEvent`.
    """
    self._generation += 1
    event = ReloadEvent(
        kind=kind,
        paths=tuple(paths),
        generation=self._generation,
    )
    for callback in list(self._callbacks):
        callback(event)
    waiters = self._waiters
    self._waiters = []
    for waiter in waiters:
        if not waiter.done():
            waiter.set_result(event)
    return event

wait async 

wait() -> ReloadEvent

Await the next reload event.

Returns:

Type Description
ReloadEvent

The next :class:ReloadEvent emitted by :meth:trigger.
Source code in tempestweb/devserver/reload.py 
149
150
151
152
153
154
155
156
157
158
async def wait(self) -> ReloadEvent:
    """Await the next reload event.

    Returns:
        The next :class:`ReloadEvent` emitted by :meth:`trigger`.
    """
    loop = asyncio.get_running_loop()
    future: asyncio.Future[ReloadEvent] = loop.create_future()
    self._waiters.append(future)
    return await future

FileWatcher

Turns filesystem changes under a project root into reload events.

The watcher filters changes by suffix, deduplicates a batch into a sorted tuple of project-relative paths, and triggers the shared :class:ReloadSignal once per batch.

Source code in tempestweb/devserver/watcher.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class FileWatcher:
    """Turns filesystem changes under a project root into reload events.

    The watcher filters changes by suffix, deduplicates a batch into a sorted
    tuple of project-relative paths, and triggers the shared
    :class:`ReloadSignal` once per batch.
    """

    def __init__(
        self,
        root: str | os.PathLike[str],
        signal: ReloadSignal,
        *,
        suffixes: Iterable[str] = DEFAULT_WATCH_SUFFIXES,
        kind: ReloadKind = ReloadKind.RESTART,
        ignore: Iterable[str | os.PathLike[str]] = (),
    ) -> None:
        """Initialize the watcher.

        Args:
            root: The project directory to watch.
            signal: The reload hub to trigger on each relevant change.
            suffixes: File suffixes that count as a reload-worthy change. A path
                whose suffix is not listed is ignored. Defaults to
                :data:`DEFAULT_WATCH_SUFFIXES`.
            kind: The reload kind emitted for every change. Defaults to
                :attr:`ReloadKind.RESTART` (clean state — the v1 behavior).
            ignore: Directories whose contents never count as a change. Used to
                exclude the build output dir (``dist/``) so a rebuild writing
                into the watched tree does not retrigger itself in a loop.
        """
        self.root: Path = Path(root).resolve()
        self.signal: ReloadSignal = signal
        self.suffixes: tuple[str, ...] = tuple(suffixes)
        self.kind: ReloadKind = kind
        self.ignore: tuple[Path, ...] = tuple(Path(p).resolve() for p in ignore)

    def _relevant(self, paths: ChangeBatch) -> tuple[str, ...]:
        """Filter a change batch to reload-worthy, project-relative paths.

        Args:
            paths: Absolute or relative paths reported as changed.

        Returns:
            A sorted, de-duplicated tuple of paths relative to :attr:`root` (or
            the original path if it is not under the root). Empty if nothing in
            the batch matched a watched suffix.
        """
        kept: set[str] = set()
        for raw in paths:
            path = Path(raw)
            if self.suffixes and path.suffix not in self.suffixes:
                continue
            resolved = path.resolve()
            if any(
                ignored == resolved or ignored in resolved.parents
                for ignored in self.ignore
            ):
                continue
            try:
                rel = resolved.relative_to(self.root)
                kept.add(str(rel))
            except ValueError:
                kept.add(str(path))
        return tuple(sorted(kept))

    def handle_batch(self, paths: ChangeBatch) -> ReloadEvent | None:
        """Process one change batch and trigger a reload if anything matched.

        Args:
            paths: The paths reported as changed in this batch.

        Returns:
            The emitted :class:`ReloadEvent`, or ``None`` when nothing in the
            batch matched a watched suffix (no reload triggered).
        """
        relevant = self._relevant(paths)
        if not relevant:
            return None
        return self.signal.trigger(kind=self.kind, paths=list(relevant))

    async def run(
        self,
        stream: ChangeStream | None = None,
        *,
        stream_factory: Callable[[Path], ChangeStream] | None = None,
    ) -> None:
        """Consume a change stream until it is exhausted, triggering reloads.

        Args:
            stream: An async iterable of change batches to consume directly.
                Mutually exclusive with ``stream_factory``. The tests pass this.
            stream_factory: A factory that builds the change stream from the
                resolved root. Defaults to a :func:`watchfiles.awatch` adapter.

        Raises:
            ValueError: If both ``stream`` and ``stream_factory`` are provided.
        """
        if stream is not None and stream_factory is not None:
            raise ValueError("pass either stream or stream_factory, not both")
        if stream is None:
            factory = stream_factory or _watchfiles_stream
            stream = factory(self.root)
        async for batch in stream:
            self.handle_batch(batch)

handle_batch 

handle_batch(paths: ChangeBatch) -> ReloadEvent | None

Process one change batch and trigger a reload if anything matched.

Parameters:

Name Type Description Default
paths ChangeBatch

The paths reported as changed in this batch.

 required

Returns:

Type Description
ReloadEvent | None

The emitted :class:ReloadEvent, or None when nothing in the
ReloadEvent | None

batch matched a watched suffix (no reload triggered).
Source code in tempestweb/devserver/watcher.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def handle_batch(self, paths: ChangeBatch) -> ReloadEvent | None:
    """Process one change batch and trigger a reload if anything matched.

    Args:
        paths: The paths reported as changed in this batch.

    Returns:
        The emitted :class:`ReloadEvent`, or ``None`` when nothing in the
        batch matched a watched suffix (no reload triggered).
    """
    relevant = self._relevant(paths)
    if not relevant:
        return None
    return self.signal.trigger(kind=self.kind, paths=list(relevant))

run async 

run(stream: ChangeStream | None = None, *, stream_factory: Callable[[Path], ChangeStream] | None = None) -> None

Consume a change stream until it is exhausted, triggering reloads.

Parameters:

Name Type Description Default
stream ChangeStream | None

An async iterable of change batches to consume directly. Mutually exclusive with stream_factory. The tests pass this.

 None
stream_factory Callable[[Path], ChangeStream] | None

A factory that builds the change stream from the resolved root. Defaults to a :func:watchfiles.awatch adapter.

 None

Raises:

Type Description
ValueError

If both stream and stream_factory are provided.
Source code in tempestweb/devserver/watcher.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def run(
    self,
    stream: ChangeStream | None = None,
    *,
    stream_factory: Callable[[Path], ChangeStream] | None = None,
) -> None:
    """Consume a change stream until it is exhausted, triggering reloads.

    Args:
        stream: An async iterable of change batches to consume directly.
            Mutually exclusive with ``stream_factory``. The tests pass this.
        stream_factory: A factory that builds the change stream from the
            resolved root. Defaults to a :func:`watchfiles.awatch` adapter.

    Raises:
        ValueError: If both ``stream`` and ``stream_factory`` are provided.
    """
    if stream is not None and stream_factory is not None:
        raise ValueError("pass either stream or stream_factory, not both")
    if stream is None:
        factory = stream_factory or _watchfiles_stream
        stream = factory(self.root)
    async for batch in stream:
        self.handle_batch(batch)

create_dev_app 

create_dev_app(out_dir: str | Path, signal: ReloadSignal | None = None) -> Starlette

Build a Starlette app that serves out_dir with optional livereload.

When signal is provided the app exposes /__livereload (an SSE stream that emits a reload event on every signal trigger) and /__livereload.js (the browser snippet), and injects the snippet's <script> tag into the served index.html. Without a signal it is a plain static host for the bundle (used by run --mode wasm).

Parameters:

Name Type Description Default
out_dir str | Path

The built artifact directory to serve at /.

 required
signal ReloadSignal | None

The reload hub to bridge to the browser. None disables livereload (static-only serving).

 None

Returns:

Type Description
Starlette

A configured :class:~starlette.applications.Starlette app.
Source code in tempestweb/devserver/http.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def create_dev_app(
    out_dir: str | Path, signal: ReloadSignal | None = None
) -> Starlette:
    """Build a Starlette app that serves ``out_dir`` with optional livereload.

    When ``signal`` is provided the app exposes ``/__livereload`` (an SSE stream
    that emits a ``reload`` event on every signal trigger) and ``/__livereload.js``
    (the browser snippet), and injects the snippet's ``<script>`` tag into the
    served ``index.html``. Without a signal it is a plain static host for the
    bundle (used by ``run --mode wasm``).

    Args:
        out_dir: The built artifact directory to serve at ``/``.
        signal: The reload hub to bridge to the browser. ``None`` disables
            livereload (static-only serving).

    Returns:
        A configured :class:`~starlette.applications.Starlette` app.
    """
    root = Path(out_dir)
    live = signal is not None

    async def index(request: Request) -> Response:
        """Serve the artifact's ``index.html``, injecting livereload in dev."""
        html = (root / "index.html").read_text(encoding="utf-8")
        return HTMLResponse(inject_livereload(html) if live else html)

    async def livereload_js(request: Request) -> Response:
        """Serve the livereload client snippet (dev only)."""
        return PlainTextResponse(_livereload_js(), media_type="application/javascript")

    async def livereload_stream(request: Request) -> Response:
        """Stream a ``reload`` SSE event on every reload signal (dev only)."""
        assert signal is not None  # guarded by `live` route registration
        return StreamingResponse(
            livereload_frames(signal),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )

    routes: list[Route | Mount] = [Route("/", index, methods=["GET"])]
    if live:
        routes.append(Route("/index.html", index, methods=["GET"]))
        routes.append(Route("/__livereload", livereload_stream, methods=["GET"]))
        routes.append(Route("/__livereload.js", livereload_js, methods=["GET"]))
    # Everything else (bundle assets, icons, sw.js, the zip) is served statically.
    routes.append(Mount("/", app=StaticFiles(directory=str(root)), name="static"))
    return Starlette(routes=routes)

livereload_frames async 

livereload_frames(signal: ReloadSignal) -> AsyncIterator[str]

Yield SSE frames for the livereload stream: open comment, then reloads.

Emits an initial ": connected" comment frame so the client knows the stream is open, then one reload event per :meth:ReloadSignal.trigger, carrying the reload generation as the SSE data. Runs until the consumer (the HTTP response) is closed.

Parameters:

Name Type Description Default
signal ReloadSignal

The reload hub to await reloads from.

 required

Yields:

Type Description
AsyncIterator[str]

SSE wire text blocks (": connected" first, then event: reload +
AsyncIterator[str]

data: <generation> blocks).
Source code in tempestweb/devserver/http.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def livereload_frames(signal: ReloadSignal) -> AsyncIterator[str]:
    """Yield SSE frames for the livereload stream: open comment, then reloads.

    Emits an initial ``": connected"`` comment frame so the client knows the
    stream is open, then one ``reload`` event per :meth:`ReloadSignal.trigger`,
    carrying the reload generation as the SSE ``data``. Runs until the consumer
    (the HTTP response) is closed.

    Args:
        signal: The reload hub to await reloads from.

    Yields:
        SSE wire text blocks (``": connected"`` first, then ``event: reload`` +
        ``data: <generation>`` blocks).
    """
    yield ": connected\n\n"
    while True:
        event = await signal.wait()
        yield f"event: reload\ndata: {event.generation}\n\n"

make_server 

make_server(app: Starlette, host: str, port: int) -> uvicorn.Server

Build a non-started uvicorn server for app bound to host:port.

Splitting construction from running lets the dev loop drive server.serve() concurrently with the file watcher under one event loop, and lets tests assert the bind config without opening a socket.

Parameters:

Name Type Description Default
app Starlette

The Starlette app to serve.

 required
host str

The bind address.

 required
port int

The bind port.

 required

Returns:

Type Description
Server

A configured (but not started) :class:uvicorn.Server.
Source code in tempestweb/devserver/http.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def make_server(app: Starlette, host: str, port: int) -> uvicorn.Server:
    """Build a non-started uvicorn server for ``app`` bound to ``host:port``.

    Splitting construction from running lets the dev loop drive ``server.serve()``
    concurrently with the file watcher under one event loop, and lets tests assert
    the bind config without opening a socket.

    Args:
        app: The Starlette app to serve.
        host: The bind address.
        port: The bind port.

    Returns:
        A configured (but not started) :class:`uvicorn.Server`.
    """
    import uvicorn

    config = uvicorn.Config(app, host=host, port=port, log_level="warning")
    return uvicorn.Server(config)

serve 

serve(app: Starlette, host: str, port: int) -> None

Serve app under uvicorn until stopped (blocking).

Parameters:

Name Type Description Default
app Starlette

The Starlette app to serve.

 required
host str

The bind address.

 required
port int

The bind port.

 required
Source code in tempestweb/devserver/http.py 
176
177
178
179
180
181
182
183
184
def serve(app: Starlette, host: str, port: int) -> None:
    """Serve ``app`` under uvicorn until stopped (blocking).

    Args:
        app: The Starlette app to serve.
        host: The bind address.
        port: The bind port.
    """
    make_server(app, host, port).run()