Coverage for src/homeassistant_repl/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 

3ha_repl_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. `hass_api` is bound the same way - an 

7already-connected homeassistant_api.AsyncClient, not a function to call. 

8 

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

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

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

12"between commands" is). 

13""" 

14 

15from __future__ import annotations 

16 

17import ast 

18import codeop 

19import sys 

20from pathlib import Path 

21 

22from prompt_toolkit import PromptSession 

23from prompt_toolkit.history import FileHistory 

24from prompt_toolkit.lexers import PygmentsLexer 

25from pygments.lexers.python import PythonLexer 

26 

27from .api_objtree import ApiObjTree, Cache 

28from .client import Client, HaReplError 

29from .local_session import LocalSession 

30from .repl import is_quit_call 

31from .rest import hass_api 

32 

33HISTORY = Path.home() / ".ha_repl_api_history" 

34 

35 

36async def run_api_repl(client: Client, ttl: float, *, auto_await: bool = True) -> int: 

37 cache = Cache(client, ttl) 

38 await cache.refresh() 

39 try: 

40 rest_client = await hass_api(client.url, client.token) 

41 except HaReplError as err: 

42 print(f"ha-repl: hass_api unavailable: {err}", file=sys.stderr) 

43 rest_client = None 

44 

45 session = LocalSession( 

46 {"obj": ApiObjTree(cache), "hass_api": rest_client}, auto_await=auto_await 

47 ) 

48 

49 compiler = codeop.CommandCompiler() 

50 compiler.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

51 prompt: PromptSession[str] = PromptSession( 

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

53 ) 

54 print( 

55 f"Home Assistant REPL (API client mode) connected to {client.url}. " 

56 "`obj` (read-only) and `hass_api` (REST client) - no `hass`. Ctrl-D to exit." 

57 ) 

58 lines: list[str] = [] 

59 while True: 

60 try: 

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

62 except KeyboardInterrupt: 

63 lines.clear() 

64 continue 

65 except EOFError: 

66 return 0 

67 lines.append(line) 

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

69 try: 

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

71 if compiler(source, "<ha_repl>", "single") is None: 

72 continue 

73 except SyntaxError, OverflowError, ValueError: 

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

75 lines.clear() 

76 if not source.strip(): 

77 continue 

78 if is_quit_call(source): 

79 return 0 

80 if cache.is_stale(): 

81 await cache.refresh() 

82 await session.run(source)