Coverage for src/dev_shell/apirepl.py: 0%

1"""API client mode's interactive shell: same local line editing/history as 

2repl.py, but code runs right here (LocalSession) - no `hass`, no 

3dev_shell_server/exec call, no HACS component required on the HA side at 

4all. `obj` is bound the same name as custom mode, just to an ApiObjTree 

5instead of the live ObjTree, so a snippet that only touches `obj` runs 

6unchanged in either mode. 

7 

8The cache refresh happens here, between prompts, not inside ApiObjTree's own 

9Mapping methods - see api_objtree.py's module docstring for why (asyncio 

10reentrancy: this loop is the one safe place that's both async and knows when 

11"between commands" is). 

12""" 

13 

14from __future__ import annotations 

15 

16import ast 

17import codeop 

18from pathlib import Path 

19 

20from prompt_toolkit import PromptSession 

21from prompt_toolkit.history import FileHistory 

22from prompt_toolkit.lexers import PygmentsLexer 

23from pygments.lexers.python import PythonLexer 

24 

25from .api_objtree import ApiObjTree, Cache 

26from .client import Client 

27from .local_session import LocalSession 

28from .repl import is_quit_call 

29 

30HISTORY = Path.home() / ".dev_shell_api_history" 

31 

32 

33async def run_api_repl(client: Client, ttl: float) -> int: 

34 cache = Cache(client, ttl) 

35 await cache.refresh() 

36 session = LocalSession({"obj": ApiObjTree(cache)}) 

37 

38 compiler = codeop.CommandCompiler() 

39 compiler.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

40 prompt: PromptSession[str] = PromptSession( 

41 history=FileHistory(str(HISTORY)), lexer=PygmentsLexer(PythonLexer) 

42 ) 

43 print( 

44 f"Dev Shell (API client mode) connected to {client.url}. " 

45 "`obj` only, read-only - no `hass`. Ctrl-D to exit." 

46 ) 

47 lines: list[str] = [] 

48 while True: 

49 try: 

50 line = await prompt.prompt_async("... " if lines else ">>> ") 

51 except KeyboardInterrupt: 

52 lines.clear() 

53 continue 

54 except EOFError: 

55 return 0 

56 lines.append(line) 

57 source = "\n".join(lines) 

58 try: 

59 # None means incomplete input: keep reading continuation lines. 

60 if compiler(source, "<dev_shell>", "single") is None: 

61 continue 

62 except SyntaxError, OverflowError, ValueError: 

63 pass # run it anyway so LocalSession reports the error consistently 

64 lines.clear() 

65 if not source.strip(): 

66 continue 

67 if is_quit_call(source): 

68 return 0 

69 if cache.is_stale(): 

70 await cache.refresh() 

71 await session.run(source)