Coverage for src/lektor_ng/admin/modules/livereload.py: 61%

1from __future__ import annotations 

2 

3import queue 

4import secrets 

5from collections.abc import Generator 

6from typing import TYPE_CHECKING 

7 

8from flask import Blueprint 

9 

10from lektor_ng.admin.utils import eventstream 

11from lektor_ng.reporter import reporter 

12 

13if TYPE_CHECKING: 

14 from lektor.builder import Artifact 

15 

16PING_DELAY = 1.0 

17 

18bp = Blueprint("livereload", __name__) 

19# Use a random ID to detect reloading, which will be changed after reloaded 

20version_id = secrets.token_urlsafe(16) 

21 

22 

23@bp.route("/events") 

24@eventstream 

25def events() -> Generator[dict[str, str]]: 

26 updated_artifacts: queue.Queue[Artifact] = queue.Queue() 

27 with reporter.on_build_change(updated_artifacts.put): 

28 while True: 

29 yield {"type": "ping", "versionId": version_id} 

30 

31 try: 

32 artifact = updated_artifacts.get(timeout=PING_DELAY) 

33 except queue.Empty: 

34 continue 

35 yield {"type": "reload", "path": artifact.artifact_name}