Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,13 @@ class ServeCommand(Command):
name = "serve"
description = "Start the FastAPI server."

# WebSocket backends accepted by uvicorn's ``--ws`` option. ``auto`` is the
# safe default: uvicorn only imports a concrete backend when a WebSocket
# connection is actually opened, so serving never requires the optional
# ``websockets`` package for apps that don't use WebSockets.
WS_BACKENDS = ("auto", "none", "websockets", "websockets-sansio", "wsproto")
DEFAULT_WS_BACKEND = "auto"

options = [
option(
"port",
Expand Down Expand Up @@ -39,6 +46,17 @@ class ServeCommand(Command):
default="bootstrap.application:app",
description="The application to serve",
),
option(
"ws",
None,
flag=False,
default=None,
description=(
"WebSocket backend passed to uvicorn: "
"auto, none, websockets, websockets-sansio, wsproto. "
"Defaults to 'auto' (overrides fastapi config)"
),
),
]

def resolve_option(self, key: str, default: str | int | None = None):
Expand All @@ -57,12 +75,21 @@ def resolve_url(self) -> Uriable:

return uri.with_port(port) if port else uri

def resolve_ws(self) -> str:
"""Select the uvicorn WebSocket backend: CLI flag > config > safe default."""
return self.option("ws") or Config.get("fastapi.ws") or self.DEFAULT_WS_BACKEND

def handle(self):
import uvicorn

from fastapi_startkit import Config
from fastapi_startkit.container import Container

ws = self.resolve_ws()
if ws not in self.WS_BACKENDS:
self.line(f"<error>Invalid --ws backend '{ws}'. Allowed values: {', '.join(self.WS_BACKENDS)}.</error>")
return 1

# Resolve server settings: CLI flag > fastapi config > uvicorn default (None)
cfg_reload_dirs = Config.get("fastapi.reload_dirs") or None
cfg_reload_excludes = Config.get("fastapi.reload_excludes") or None
Expand All @@ -74,7 +101,7 @@ def handle(self):
"host": url.host(),
"port": url.port(),
"reload": reload,
"ws": "websockets-sansio",
"ws": ws,
}

if self.is_app_exist():
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@
class FastAPIConfig:
app_url: str = dataclasses.field(default_factory=lambda: env("APP_URL", "http://127.0.0.1:8000"))
reload: bool = dataclasses.field(default_factory=lambda: env("APP_RELOAD", True))
# uvicorn WebSocket backend: auto, none, websockets, websockets-sansio, wsproto.
# 'auto' never requires the optional 'websockets' package unless a WS connection is opened.
ws: str = dataclasses.field(default_factory=lambda: env("APP_WS", "auto"))
reload_dirs: list | None = None
reload_excludes: list = dataclasses.field(
default_factory=lambda: [
Expand Down
52 changes: 51 additions & 1 deletion fastapi_startkit/tests/fastapi/test_serve_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,11 +85,61 @@ def test_default_port_in_output(self):
tester, _ = run()
assert str(_DEFAULT_PORT) in tester.io.fetch_output()

def test_uvicorn_kwargs_contain_ws(self):
def test_uvicorn_kwargs_default_ws_is_auto(self):
"""The default WebSocket backend must be the safe 'auto', never
'websockets-sansio' — otherwise serve crashes when the optional
'websockets' package is not installed.
"""
_, mock_uvicorn = run()
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "auto"


# ---------------------------------------------------------------------------
# 6. --ws option — WebSocket backend selection
# ---------------------------------------------------------------------------


class TestWsOption:
def test_ws_flag_passed_to_uvicorn(self):
_, mock_uvicorn = run("--ws websockets")
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "websockets"

def test_ws_websockets_sansio_opt_in(self):
_, mock_uvicorn = run("--ws websockets-sansio")
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "websockets-sansio"

def test_ws_none_backend(self):
_, mock_uvicorn = run("--ws none")
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "none"

def test_ws_config_used_when_no_cli_flag(self):
_, mock_uvicorn = run(config={"fastapi.ws": "wsproto"})
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "wsproto"

def test_cli_flag_overrides_config(self):
_, mock_uvicorn = run("--ws auto", config={"fastapi.ws": "wsproto"})
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "auto"

def test_invalid_ws_exits_nonzero(self):
tester, _ = run("--ws bogus")
assert tester.status_code == 1

def test_invalid_ws_does_not_call_uvicorn(self):
_, mock_uvicorn = run("--ws bogus")
mock_uvicorn.assert_not_called()

def test_invalid_ws_reports_allowed_values(self):
tester, _ = run("--ws bogus")
output = tester.io.fetch_output() + tester.io.fetch_error()
assert "bogus" in output
assert "websockets-sansio" in output


# ---------------------------------------------------------------------------
# 2. CLI --host / --port override defaults
Expand Down
2 changes: 1 addition & 1 deletion fastapi_startkit/uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading