Coverage for src / beaverbunch / network / app.py: 100.0%
124 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 00:24 +0000
1"""FastAPI HTTP transport layer for the Beaverbunch lobby server.
3Every endpoint simply converts the HTTP request body into the canonical
4protocol dict, forwards it to ``SessionServer.handle()``, and returns the
5resulting response dict as JSON. All real logic lives in
6:mod:`beaverbunch.network.server` / :mod:`beaverbunch.network.session`.
7"""
9from __future__ import annotations
11from typing import Any
13from fastapi import FastAPI, Request
14from fastapi.middleware.cors import CORSMiddleware
15from pydantic import BaseModel
17from beaverbunch import __version__
18from beaverbunch.network.config import RuntimeConfig, load_runtime_config
19from beaverbunch.network.server import SessionServer
20from beaverbunch.network.session import SessionManager
23# ---------------------------------------------------------------------------
24# App factory helpers
25# ---------------------------------------------------------------------------
27def _get_server(request: Request) -> SessionServer:
28 return request.app.state.session_server
31def _dispatch(request: Request, message: dict[str, Any]) -> dict[str, Any]:
32 return _get_server(request).handle(message)
35def create_app(config: RuntimeConfig | None = None) -> FastAPI:
36 runtime_config = config or load_runtime_config()
37 application = FastAPI(title=runtime_config.app_name, version=runtime_config.app_version)
38 application.state.runtime_config = runtime_config
39 application.state.session_server = SessionServer(
40 manager=SessionManager(max_sessions=runtime_config.max_sessions),
41 )
43 if runtime_config.cors_origins:
44 application.add_middleware(
45 CORSMiddleware,
46 allow_origins=list(runtime_config.cors_origins),
47 allow_methods=["*"],
48 allow_headers=["*"],
49 )
51 # -----------------------------------------------------------------------
52 # Operational endpoints
53 # -----------------------------------------------------------------------
55 @application.get("/health", summary="Liveness probe")
56 def health() -> dict[str, Any]:
57 return {
58 "status": "ok",
59 "service": "beaverbunch-backend",
60 "version": __version__,
61 }
63 @application.get("/ready", summary="Readiness probe")
64 def ready(request: Request) -> dict[str, Any]:
65 server = _get_server(request)
66 config = request.app.state.runtime_config
67 return {
68 "status": "ready",
69 "sessions": {
70 "active": server.manager.session_count,
71 "capacity": config.max_sessions,
72 },
73 }
75 # -----------------------------------------------------------------------
76 # Lobby endpoints
77 # -----------------------------------------------------------------------
79 @application.post("/sessions", summary="Create a new lobby session", status_code=201)
80 def create_session(request: Request, body: CreateSessionBody) -> dict[str, Any]:
81 message: dict[str, Any] = {"type": "create_session", "host_name": body.host_name}
82 if body.settings is not None:
83 message["settings"] = body.settings
84 return _dispatch(request, message)
86 @application.post("/sessions/{join_code}/join", summary="Join an existing session")
87 def join_session(request: Request, join_code: str, body: JoinSessionBody) -> dict[str, Any]:
88 return _dispatch(
89 request,
90 {
91 "type": "join_session",
92 "join_code": join_code,
93 "player_name": body.player_name,
94 },
95 )
97 @application.post("/sessions/{join_code}/start", summary="Start the game (host only)")
98 def start_session(request: Request, join_code: str, body: StartSessionBody) -> dict[str, Any]:
99 return _dispatch(
100 request,
101 {
102 "type": "start_session",
103 "join_code": join_code,
104 "player_token": body.player_token,
105 },
106 )
108 @application.get("/sessions/{join_code}", summary="Fetch current session snapshot")
109 def get_session(request: Request, join_code: str) -> dict[str, Any]:
110 return _dispatch(
111 request,
112 {
113 "type": "get_session",
114 "join_code": join_code,
115 },
116 )
118 @application.post("/sessions/{join_code}/leave", summary="Leave a session")
119 def leave_session(request: Request, join_code: str, body: LeaveSessionBody) -> dict[str, Any]:
120 return _dispatch(
121 request,
122 {
123 "type": "leave_session",
124 "join_code": join_code,
125 "player_token": body.player_token,
126 },
127 )
129 @application.post("/sessions/{join_code}/close", summary="Close a session (host only)")
130 def close_session(request: Request, join_code: str, body: CloseSessionBody) -> dict[str, Any]:
131 return _dispatch(
132 request,
133 {
134 "type": "close_session",
135 "join_code": join_code,
136 "player_token": body.player_token,
137 },
138 )
140 # -----------------------------------------------------------------------
141 # Game-phase endpoints
142 # -----------------------------------------------------------------------
144 @application.get("/sessions/{join_code}/game", summary="Get current game state (perspective-aware)")
145 def get_game(request: Request, join_code: str, player_token: str) -> dict[str, Any]:
146 return _dispatch(
147 request,
148 {
149 "type": "get_game",
150 "join_code": join_code,
151 "player_token": player_token,
152 },
153 )
155 @application.post("/sessions/{join_code}/game/peek-initial", summary="Peek at initial cards (game start)")
156 def peek_initial(request: Request, join_code: str, body: PeekInitialBody) -> dict[str, Any]:
157 return _dispatch(
158 request,
159 {
160 "type": "peek_initial",
161 "join_code": join_code,
162 "player_token": body.player_token,
163 "indices": body.indices,
164 },
165 )
167 @application.post("/sessions/{join_code}/game/draw", summary="Draw a card from the deck")
168 def draw_card(request: Request, join_code: str, body: DrawCardBody) -> dict[str, Any]:
169 return _dispatch(
170 request,
171 {
172 "type": "draw_card",
173 "join_code": join_code,
174 "player_token": body.player_token,
175 },
176 )
178 @application.post("/sessions/{join_code}/game/draw-discard", summary="Draw the top card from the discard pile")
179 def draw_discard(request: Request, join_code: str, body: DrawDiscardBody) -> dict[str, Any]:
180 return _dispatch(
181 request,
182 {
183 "type": "draw_discard",
184 "join_code": join_code,
185 "player_token": body.player_token,
186 },
187 )
189 @application.post("/sessions/{join_code}/game/keep", summary="Keep drawn card, replacing a hand card")
190 def keep_card(request: Request, join_code: str, body: KeepCardBody) -> dict[str, Any]:
191 return _dispatch(
192 request,
193 {
194 "type": "keep_card",
195 "join_code": join_code,
196 "player_token": body.player_token,
197 "hand_index": body.hand_index,
198 },
199 )
201 @application.post("/sessions/{join_code}/game/discard", summary="Discard the drawn card")
202 def discard_drawn(request: Request, join_code: str, body: DiscardDrawnBody) -> dict[str, Any]:
203 return _dispatch(
204 request,
205 {
206 "type": "discard_drawn",
207 "join_code": join_code,
208 "player_token": body.player_token,
209 },
210 )
212 @application.post("/sessions/{join_code}/game/peek", summary="Jack bonus: peek at own card")
213 def peek_own(request: Request, join_code: str, body: PeekOwnBody) -> dict[str, Any]:
214 return _dispatch(
215 request,
216 {
217 "type": "peek_own",
218 "join_code": join_code,
219 "player_token": body.player_token,
220 "hand_index": body.hand_index,
221 },
222 )
224 @application.post(
225 "/sessions/{join_code}/game/swap", summary="Queen bonus: swap own card with another player's card",
226 )
227 def swap_card(request: Request, join_code: str, body: SwapCardBody) -> dict[str, Any]:
228 return _dispatch(
229 request,
230 {
231 "type": "swap_card",
232 "join_code": join_code,
233 "player_token": body.player_token,
234 "own_index": body.own_index,
235 "target_player_name": body.target_player_name,
236 "target_index": body.target_index,
237 },
238 )
240 @application.post("/sessions/{join_code}/game/skip-bonus", summary="Skip the current bonus action")
241 def skip_bonus(request: Request, join_code: str, body: SkipBonusBody) -> dict[str, Any]:
242 return _dispatch(
243 request,
244 {
245 "type": "skip_bonus",
246 "join_code": join_code,
247 "player_token": body.player_token,
248 },
249 )
251 @application.post("/sessions/{join_code}/game/snap", summary="Snap: claim a hand card matches the top discard")
252 def snap(request: Request, join_code: str, body: SnapBody) -> dict[str, Any]:
253 return _dispatch(
254 request,
255 {
256 "type": "snap",
257 "join_code": join_code,
258 "player_token": body.player_token,
259 "hand_index": body.hand_index,
260 },
261 )
263 @application.post("/sessions/{join_code}/game/beaver", summary="Declare Beaver to trigger the last round")
264 def beaver(request: Request, join_code: str, body: BeaverBody) -> dict[str, Any]:
265 return _dispatch(
266 request,
267 {
268 "type": "beaver",
269 "join_code": join_code,
270 "player_token": body.player_token,
271 },
272 )
274 return application
277# ---------------------------------------------------------------------------
278# Pydantic request bodies
279# ---------------------------------------------------------------------------
281class CreateSessionBody(BaseModel):
282 host_name: str | None = None
283 settings: dict[str, Any] | None = None
286class JoinSessionBody(BaseModel):
287 player_name: str
290class StartSessionBody(BaseModel):
291 player_token: str
294class LeaveSessionBody(BaseModel):
295 player_token: str
298class CloseSessionBody(BaseModel):
299 player_token: str
302class GetGameBody(BaseModel):
303 player_token: str
306class PeekInitialBody(BaseModel):
307 player_token: str
308 indices: list[int]
311class DrawCardBody(BaseModel):
312 player_token: str
315class DrawDiscardBody(BaseModel):
316 player_token: str
319class KeepCardBody(BaseModel):
320 player_token: str
321 hand_index: int
324class DiscardDrawnBody(BaseModel):
325 player_token: str
328class PeekOwnBody(BaseModel):
329 player_token: str
330 hand_index: int
333class SwapCardBody(BaseModel):
334 player_token: str
335 own_index: int
336 target_player_name: str
337 target_index: int
340class SkipBonusBody(BaseModel):
341 player_token: str
344class SnapBody(BaseModel):
345 player_token: str
346 hand_index: int
349class BeaverBody(BaseModel):
350 player_token: str
353app = create_app()