Coverage for frappe_manager / commands / logs.py: 48%

1from typing import Annotated 

2 

3import typer 

4from typer_examples import example 

5 

6from frappe_manager.commands import check_bench_migration_required 

7from frappe_manager.output_manager import get_global_output_handler 

8from frappe_manager.site_manager.site import Bench 

9from frappe_manager.utils.callbacks import ( 

10 sitename_callback, 

11 sites_autocompletion_callback, 

12) 

13 

14 

15@example( 

16 "Show frappe server logs", 

17 "{benchname}", 

18 detail="Displays the frappe service logs for the bench; useful to inspect server output and errors.", 

19 benchname="mybench", 

20) 

21@example( 

22 "Follow logs in real-time", 

23 "{benchname} -f", 

24 detail="Streams logs continuously; press Ctrl+C to stop following.", 

25 benchname="mybench", 

26) 

27@example( 

28 "Show nginx container logs", 

29 "{benchname} --service nginx -f", 

30 detail="Shows the nginx container logs for the bench and follows them in real-time when -f is provided.", 

31 benchname="mybench", 

32) 

33@example( 

34 "Show redis logs", 

35 "{benchname} --service redis-cache", 

36 detail="Displays logs from redis-cache service; helpful when debugging caching or queuing issues.", 

37 benchname="mybench", 

38) 

39def logs( 

40 ctx: typer.Context, 

41 benchname: Annotated[ 

42 str | None, 

43 typer.Argument( 

44 help="Name of the bench.", 

45 autocompletion=sites_autocompletion_callback, 

46 callback=sitename_callback, 

47 ), 

48 ] = None, 

49 service: Annotated[str | None, typer.Option(help="Service name (frappe, nginx, redis-cache, etc.)")] = None, 

50 follow: Annotated[bool, typer.Option("--follow", "-f", help="Follow logs in real-time")] = False, 

51): 

52 """ 

53 Show bench logs (server or container). 

54 

55 View logs from bench services (frappe, nginx, redis, etc.) with optional follow mode. 

56 """ 

57 

58 check_bench_migration_required(benchname) 

59 

60 services_manager = ctx.obj["services"] 

61 verbose = ctx.obj["verbose"] 

62 

63 output = get_global_output_handler() 

64 logger = ctx.obj.get("logger") 

65 bench = Bench.get_object(benchname, services_manager, logger=logger, output_handler=output) 

66 

67 if service: 

68 available_services = bench.get_available_services() 

69 if service not in available_services: 

70 output.display_error(f"Service '{service}' not found") 

71 output.print(f"Available services: {', '.join(sorted(available_services))}") 

72 raise typer.Exit(1) 

73 

74 bench.logs(follow, service)