Coverage for core / src / sensorkit / cli / utils.py: 0%

1# SPDX-License-Identifier: Apache-2.0 

2import functools 

3from contextlib import contextmanager 

4from typing import Any, Callable 

5 

6import asyncclick as click 

7from rich.console import Console 

8 

9from sensorkit.api.entrypoint import ShutdownSignal 

10 

11console = Console() 

12 

13 

14def common_options(f: Callable[..., Any]) -> Callable[..., Any]: 

15 """Decorator to add common SensorKit options to a command.""" 

16 # Placeholder for common options like --config, --log-level etc. 

17 # For now, we can just return f 

18 return f 

19 

20 

21def entity_option(default: str | None = None, required: bool = False, help: str = "Entity name"): 

22 """Standardized entity option.""" 

23 return click.option("-e", "--entity", default=default, required=required, help=help) 

24 

25 

26@contextmanager 

27def report_errors(): 

28 from sensorkit.backend.base import BackendError, KVError 

29 from sensorkit.backend.lease import LeaseUnavailableError 

30 

31 try: 

32 yield 

33 except* ShutdownSignal: 

34 # Fall through to normal shutdown. 

35 pass 

36 except* KVError as eg: 

37 console.print(f"[red]Error:[/red] Configuration or state error: {eg.exceptions[0]}") 

38 raise click.Abort() from eg 

39 except* LeaseUnavailableError as eg: 

40 console.print("[red]Error:[/red] Service or entity is already running!") 

41 raise click.Abort() from eg 

42 except* BackendError as eg: 

43 console.print(f"[red]Error:[/red] Backend unavailable: {eg.exceptions[0]}") 

44 raise click.Abort() from eg 

45 except* Exception as eg: 

46 for e in eg.exceptions: 

47 console.print(f"[red]Unexpected {type(e).__name__}[/red]") 

48 

49 console.print_exception() 

50 raise click.Abort() from eg 

51 

52 

53def with_kit(f: Callable[..., Any]) -> Callable[..., Any]: 

54 """Decorator that provides a 'kit' instance to the command and handles errors.""" 

55 

56 @functools.wraps(f) 

57 async def wrapper(*args, **kwargs): 

58 from sensorkit.api.bootstrap import connect 

59 

60 with report_errors(): 

61 kit = await connect() 

62 return await f(kit, *args, **kwargs) 

63 

64 return wrapper