Coverage for seedboxsync/front/apis/logs.py: 91%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6"""SeedboxSync API Logs view.""" 

7 

8from datetime import timedelta 

9from typing import Any 

10from flask_restx import Namespace, fields, reqparse 

11from seedboxsync.core.database.models import Log 

12from seedboxsync.front.apis import Resource, parser_pagination, parser_period 

13from seedboxsync.front.login_manager import admin_required 

14 

15api = Namespace("logs", description="View logs information") 

16 

17 

18# ========================== 

19# Models 

20# ========================== 

21log_model = api.model( 

22 "Log", 

23 { 

24 "id": fields.Integer( 

25 required=True, 

26 description="Unique identifier of the log record", 

27 example=999, 

28 ), 

29 "level": fields.String( 

30 required=True, 

31 description="Severity level of the record (e.g. DEBUG, INFO, WARNING, ERROR, CRITICAL)", 

32 example="ConvallisMorbi.doc", 

33 ), 

34 "logger": fields.String( 

35 required=True, 

36 description="Name of the logger that emitted the record", 

37 example="doc", 

38 ), 

39 "module": fields.String( 

40 required=True, 

41 description="Module of the logger that emitted the record", 

42 example="doc", 

43 ), 

44 "message": fields.String( 

45 required=True, 

46 description="Formatted message recorded by the logger", 

47 example="ConvallisMorbi.doc", 

48 ), 

49 "timestamp": fields.DateTime( 

50 dt_format="iso8601", 

51 required=True, 

52 description="Date and time when the log record was created", 

53 ), 

54 }, 

55) 

56log_list_envelope = Resource.build_envelope_model(api, "LogList", nested_model=log_model) 

57log_envelope = Resource.build_envelope_model(api, "Log", nested_model=log_model, as_list=False) 

58log_message_envelope = Resource.build_envelope_model(api, "LogMessage", as_message=True) 

59 

60 

61# ========================== 

62# Request parser 

63# ========================== 

64parser = reqparse.RequestParser() 

65parser.add_argument("search", type=str, required=False, help="Optional search string to filter items") 

66parser.add_argument("level", choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"), required=False, help="Optional level to filter items") 

67parser.args.extend(parser_period.args) 

68parser.args.extend(parser_pagination.args) 

69 

70 

71# ========================== 

72# Endpoints 

73# ========================== 

74@api.route("") 

75class Logs(Resource): 

76 """Endpoint for managing log viewing.""" 

77 

78 @api.doc("list_logs") # type: ignore[untyped-decorator] 

79 @api.expect(parser) # type: ignore[untyped-decorator] 

80 @api.marshal_with(log_list_envelope, code=200, description="List of logs") # type: ignore[untyped-decorator] 

81 @admin_required 

82 def get(self) -> dict[str, Any]: 

83 """ 

84 Retrieve a list of recent logs. 

85 

86 Query Parameters: 

87 - offset: Number of items to skip before starting to collect the result set (default: 0) 

88 - limit: Maximum number of logs to return (default=50) 

89 - search: Optional search string to filter items 

90 """ 

91 args = parser.parse_args() 

92 offset = args.get("offset") 

93 limit = self.set_limit(args.get("limit", 50)) 

94 search = args.get("search") 

95 level = args.get("level") 

96 start_date = args.get("start_date") 

97 end_date = args.get("end_date") 

98 

99 count = Log.select() 

100 select = ( 

101 Log.select( 

102 Log.id, 

103 Log.level, 

104 Log.logger, 

105 Log.module, 

106 Log.message, 

107 Log.timestamp, 

108 ) 

109 .limit(limit) 

110 .offset(offset) 

111 .order_by(Log.timestamp.desc()) 

112 ) 

113 

114 if search: 

115 count = count.where(Log.message.contains(search) | Log.level.contains(search) | Log.module.contains(search)) 

116 select = select.where(Log.message.contains(search) | Log.level.contains(search) | Log.module.contains(search)) 

117 

118 if start_date: 

119 count = count.where(Log.timestamp >= start_date) 

120 select = select.where(Log.timestamp >= start_date) 

121 

122 if level: 

123 count = count.where(Log.level == level) 

124 select = select.where(Log.level == level) 

125 

126 if end_date: 

127 end_exclusive = end_date + timedelta(days=1) 

128 count = count.where(Log.timestamp < end_exclusive) 

129 select = select.where(Log.timestamp < end_exclusive) 

130 

131 return self.build_envelope(list(select.dicts()), data_total=count.count(), type="LogList")