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

1"""Crawl route handler.""" 

2 

3from __future__ import annotations 

4 

5import anyio 

6from litestar import post 

7from litestar.exceptions import ValidationException 

8from litestar.response import Stream 

9 

10from lilbee.server import handlers 

11from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

12from lilbee.server.models import CrawlLoginBody, CrawlRequest 

13 

14 

15@post("/api/crawl", media_type=SSE_MEDIA_TYPE) 

16async def crawl_route(data: CrawlRequest) -> Stream: 

17 """Crawl a URL with streaming SSE progress events.""" 

18 from lilbee.crawler import CrawlLogin, require_valid_crawl_url 

19 from lilbee.crawler.stored_login import crawl_login 

20 

21 body = data.login if data.login is not None else CrawlLoginBody() 

22 try: 

23 login = crawl_login([data.url], CrawlLogin(**body.model_dump())) 

24 except ValueError as exc: 

25 raise ValidationException(str(exc)) from exc 

26 try: 

27 # URL validation resolves the host (blocking DNS), so it runs off the loop, 

28 # mirroring the MCP crawl tool so neither async transport stalls the loop. 

29 await anyio.to_thread.run_sync(require_valid_crawl_url, data.url) 

30 except ValueError as exc: 

31 raise ValidationException(str(exc)) from exc 

32 gen = handlers.crawl_stream( 

33 url=data.url, 

34 depth=data.depth, 

35 max_pages=data.max_pages, 

36 render_mode=data.render_mode, 

37 include_subdomains=data.include_subdomains, 

38 login=login, 

39 ) 

40 return Stream(gen, media_type=SSE_MEDIA_TYPE)