Coverage for src/lektor_ng/devserver.py: 36%

1from __future__ import annotations 

2 

3import logging 

4import threading 

5import traceback 

6import webbrowser 

7from contextlib import ExitStack 

8from typing import TYPE_CHECKING, NamedTuple 

9 

10from werkzeug.serving import is_running_from_reloader, run_simple 

11 

12from lektor_ng.admin import WebAdmin 

13from lektor_ng.builder import Builder 

14from lektor_ng.db import Database 

15from lektor_ng.reporter import CliReporter 

16from lektor_ng.utils import process_extra_flags 

17from lektor_ng.watcher import watch_project 

18 

19if TYPE_CHECKING: 

20 from _typeshed import StrPath 

21 from lektor.environment import Environment 

22 

23 

24class BackgroundBuilder(threading.Thread): 

25 """Run a thread to watch the project tree and rebuild when changes are noticed. 

26 

27 This is a contextmanager. On entry, the watcher thread is started, on exit it is 

28 stopped. 

29 

30 """ 

31 

32 def __init__( 

33 self, 

34 env: Environment, 

35 output_path: StrPath, 

36 prune: bool = True, 

37 verbosity: int = 0, 

38 extra_flags: dict[str, str] | None = None, 

39 ): 

40 threading.Thread.__init__(self) 

41 self.env = env 

42 self.output_path = output_path 

43 self.prune = prune 

44 self.verbosity = verbosity 

45 self.extra_flags = extra_flags 

46 

47 # See https://github.com/samuelcolvin/watchfiles/pull/132 

48 self.stop_event = threading.Event() 

49 

50 def __enter__(self) -> BackgroundBuilder: 

51 self.start() 

52 return self 

53 

54 def __exit__(self, *args: object) -> None: 

55 self.stop_event.set() 

56 

57 def build(self, update_source_info_first: bool = False) -> None: 

58 try: 

59 db = Database(self.env) 

60 builder = Builder(db.new_pad(), self.output_path, extra_flags=self.extra_flags) 

61 if update_source_info_first: 

62 builder.update_all_source_infos() 

63 builder.build_all() 

64 if self.prune: 

65 builder.prune() 

66 except Exception: 

67 traceback.print_exc() 

68 

69 def run(self) -> None: 

70 watch = watch_project(self.env, self.output_path, stop_event=self.stop_event) 

71 with CliReporter(self.env, verbosity=self.verbosity): 

72 self.build(update_source_info_first=True) 

73 for _changes in watch: 

74 self.build() 

75 

76 

77class BindAddr(NamedTuple): 

78 host: str 

79 port: int 

80 

81 

82def browse_to_address(addr: BindAddr) -> None: 

83 timer = threading.Timer(1.0, webbrowser.open, (f"http://{addr.host}:{addr.port}",)) 

84 timer.daemon = True 

85 timer.start() 

86 

87 

88def run_server( 

89 bindaddr: BindAddr, 

90 env: Environment, 

91 output_path: StrPath, 

92 *, 

93 prune: bool = True, 

94 verbosity: int = 0, 

95 lektor_dev: bool = False, 

96 ui_lang: str = "en", 

97 browse: bool = False, 

98 extra_flags: dict[str, str] | None = None, 

99) -> None: 

100 """This runs a server but also spawns a background process. It's 

101 not safe to call this more than once per python process! 

102 """ 

103 bindaddr = BindAddr._make(bindaddr) 

104 

105 in_main_process = is_running_from_reloader() or not lektor_dev 

106 extra_flags = process_extra_flags(extra_flags) 

107 if lektor_dev: 

108 env.jinja_env.add_extension("jinja2.ext.debug") 

109 else: 

110 logging.getLogger("werkzeug").setLevel(logging.WARNING) 

111 

112 app = WebAdmin( 

113 env, 

114 output_path=output_path, 

115 verbosity=verbosity, 

116 debug=lektor_dev, 

117 ui_lang=ui_lang, 

118 extra_flags=extra_flags, 

119 ) 

120 

121 if browse and not is_running_from_reloader(): 

122 browse_to_address(bindaddr) 

123 

124 with ExitStack() as stack: 

125 if in_main_process: 

126 env.plugin_controller.emit("server-spawn", bindaddr=bindaddr, extra_flags=extra_flags) 

127 stack.callback(env.plugin_controller.emit, "server-stop") 

128 

129 background_builder = BackgroundBuilder( 

130 env, 

131 output_path=output_path, 

132 prune=prune, 

133 verbosity=verbosity, 

134 extra_flags=extra_flags, 

135 ) 

136 stack.enter_context(background_builder) 

137 

138 run_simple( 

139 bindaddr.host, 

140 bindaddr.port, 

141 app, 

142 use_debugger=True, 

143 threaded=True, 

144 use_reloader=lektor_dev, 

145 )