Skip to content
Merged
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 @@ -11,6 +11,9 @@
if TYPE_CHECKING:
from fastapi_startkit.application import Application

# Values accepted by uvicorn's `ws` setting.
WS_BACKENDS = ("auto", "none", "websockets", "websockets-sansio", "wsproto")


class ServeCommand(Command):
name = "serve"
Expand Down Expand Up @@ -45,6 +48,13 @@ class ServeCommand(Command):
default=None,
description="The application to serve (overrides fastapi config)",
),
option(
"ws",
None,
flag=False,
default=None,
description=f"The uvicorn WebSocket backend: {', '.join(WS_BACKENDS)} (overrides fastapi config)",
),
]

def config_value(self, key: str) -> Any:
Expand Down Expand Up @@ -72,11 +82,20 @@ def resolve_url(self) -> Uriable:

return uri.with_port(port) if port else uri

def resolve_ws(self) -> str:
"""CLI flag > fastapi config > FastAPIConfig default ('auto')."""
return str(self.option("ws") or self.config_value("ws"))

def handle(self) -> int:
import uvicorn

from fastapi_startkit.container import Container

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

url = self.resolve_url()
reload = self.resolve_option("reload")
app = self.resolve_option("app")
Expand All @@ -85,7 +104,7 @@ def handle(self) -> int:
"host": url.host(),
"port": url.port(),
"reload": reload,
"ws": "websockets-sansio",
"ws": ws,
}

if self.is_app_exist(app):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ class FastAPIConfig:
app_url: str = dataclasses.field(default_factory=lambda: env("APP_URL", "http://127.0.0.1:8000"))
app: str = "bootstrap.application:app"
reload: bool = dataclasses.field(default_factory=lambda: env("APP_RELOAD", True))
# 'auto' keeps the optional `websockets` package from being required at startup.
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
66 changes: 64 additions & 2 deletions fastapi_startkit/tests/fastapi/test_serve_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,10 +88,14 @@ 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") == "websockets-sansio"
assert kwargs.get("ws") == "auto"


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -325,3 +329,61 @@ def test_reload_defaults_true_without_env(self):
_, mock_uvicorn = run(app_found=True)
_, kwargs = mock_uvicorn.call_args
assert kwargs.get("reload") is True


# ---------------------------------------------------------------------------
# 9. --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

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

def test_ws_env_applied_after_import(self):
with patch.dict(os.environ, {"APP_WS": "wsproto"}):
_, mock_uvicorn = run()

_, kwargs = mock_uvicorn.call_args
assert kwargs.get("ws") == "wsproto"
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