Coverage for src / lilbee / server / routes / setup.py: 100%

1"""Setup routes: status and bootstrap for optional runtime components. 

2 

3Currently exposes Playwright Chromium bootstrap (needed for /crawl). The 

4SSE event sequence mirrors what the TUI does in 

5``TaskBarController.ensure_chromium`` so a stream consumer can render a 

6matching ``setup`` progress indicator. 

7 

8Endpoints: 

9 GET /setup/crawler/status → { installed, component, browsers_path } 

10 POST /setup/crawler → text/event-stream of setup_start → 

11 setup_progress → setup_done → done 

12""" 

13 

14from __future__ import annotations 

15 

16import asyncio 

17from collections.abc import AsyncGenerator 

18from typing import Any 

19 

20from litestar import get, post 

21from litestar.response import Stream 

22 

23from lilbee.crawler import ( 

24 bootstrap_chromium, 

25 chromium_installed, 

26 crawler_available, 

27 crawler_browsers_path, 

28) 

29from lilbee.server.handlers import SseStream, sse_done, sse_error 

30 

31 

32@get("/setup/crawler/status") 

33async def setup_crawler_status_route() -> dict[str, Any]: 

34 """Return whether the crawler is fully ready (Python package + Chromium).""" 

35 package_installed = crawler_available() 

36 chromium_ok = chromium_installed() 

37 return { 

38 "installed": package_installed and chromium_ok, 

39 "package_installed": package_installed, 

40 "chromium_installed": chromium_ok, 

41 "component": "chromium", 

42 "browsers_path": str(crawler_browsers_path()), 

43 } 

44 

45 

46async def _bootstrap_crawler_stream() -> AsyncGenerator[str, None]: 

47 sse = SseStream() 

48 

49 async def _run() -> None: 

50 try: 

51 await bootstrap_chromium(on_progress=sse.callback) 

52 finally: 

53 sse.queue.put_nowait(None) 

54 

55 task = asyncio.create_task(_run()) 

56 async for event in sse.drain(task, "Crawler setup stream"): 

57 yield event 

58 if task.done() and not task.cancelled(): 

59 exc = task.exception() 

60 if exc is not None: 

61 yield sse_error(str(exc)) 

62 return 

63 yield sse_done({}) 

64 

65 

66@post("/setup/crawler") 

67async def setup_crawler_route() -> Stream: 

68 """Stream the Chromium bootstrap subprocess as SSE events.""" 

69 return Stream(_bootstrap_crawler_stream(), media_type="text/event-stream") 

70 

71 

72__all__ = [ 

73 "setup_crawler_route", 

74 "setup_crawler_status_route", 

75]