Coverage for src/lexigram/admin/events/__init__.py: 100%

137 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-21 14:56 +0800

1"""Canonical admin event surface for lexigram-admin. 

2 

3This module is the single import point for all admin events. 

4 

5Domain-level resource and bulk-operation events (``ResourceCreated``, 

6``ResourceUpdated``, ``ResourceDeleted``, ``BulkOperationCompleted``, 

7``UserCreated``, ``UserUpdated``, ``UserDeactivated``, ``UserDeleted``) 

8are defined here. 

9 

10Operational admin events (audit, auth, export/import, system) are also 

11defined here as the ``AdminEvent`` hierarchy. 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass, field 

17from typing import Any 

18 

19from lexigram.contracts.domain.events import DomainEvent 

20 

21__all__ = [ 

22 # Action 

23 "ActionExecuted", 

24 # Admin base 

25 "AdminEvent", 

26 # System 

27 "AdminStarted", 

28 "AdminStopped", 

29 # Auth 

30 "AdminUserCreated", 

31 "AdminUserLoggedIn", 

32 "AdminUserLoggedOut", 

33 "AdminUserUpdated", 

34 # Bulk 

35 "BulkOperationCompleted", 

36 # Settings 

37 "ConfigReloaded", 

38 # Export / Import 

39 "ExportCompleted", 

40 "ExportFailed", 

41 "ExportStarted", 

42 "ImportCompleted", 

43 "ImportStarted", 

44 "PasswordChanged", 

45 # Resource 

46 "ResourceCreated", 

47 "ResourceDeleted", 

48 "ResourceRestored", 

49 "ResourceUpdated", 

50 "ResourceViewed", 

51 "RolesAssigned", 

52 "SettingsUpdated", 

53 # Domain user events 

54 "UserCreated", 

55 "UserDeactivated", 

56 "UserDeleted", 

57 "UserUpdated", 

58] 

59 

60 

61# ========== Domain-level resource events (previously domain/events.py) ========== 

62 

63 

64@dataclass(frozen=True, kw_only=True) 

65class UserCreated(DomainEvent): 

66 """Raised when an admin user is created.""" 

67 

68 user_id: str 

69 email: str 

70 

71 

72@dataclass(frozen=True, kw_only=True) 

73class UserUpdated(DomainEvent): 

74 """Raised when an admin user's data is updated.""" 

75 

76 user_id: str 

77 changes: dict[str, Any] = field(default_factory=dict) 

78 

79 

80@dataclass(frozen=True, kw_only=True) 

81class UserDeactivated(DomainEvent): 

82 """Raised when an admin user is deactivated.""" 

83 

84 user_id: str 

85 

86 

87@dataclass(frozen=True, kw_only=True) 

88class UserDeleted(DomainEvent): 

89 """Raised when an admin user is soft-deleted.""" 

90 

91 user_id: str 

92 

93 

94@dataclass(frozen=True, kw_only=True) 

95class ResourceCreated(DomainEvent): 

96 """Raised when an admin resource is created.""" 

97 

98 resource_type: str = "" 

99 resource_id: Any = None 

100 data: dict[str, Any] = field(default_factory=dict) 

101 correlation_id: str | None = None 

102 

103 

104@dataclass(frozen=True, kw_only=True) 

105class ResourceUpdated(DomainEvent): 

106 """Raised when an admin resource is updated.""" 

107 

108 resource_type: str = "" 

109 resource_id: Any = None 

110 changes: dict[str, tuple[Any, Any]] = field(default_factory=dict) 

111 correlation_id: str | None = None 

112 

113 

114@dataclass(frozen=True, kw_only=True) 

115class ResourceDeleted(DomainEvent): 

116 """Raised when an admin resource is deleted.""" 

117 

118 resource_type: str = "" 

119 resource_id: Any = None 

120 soft_delete: bool = True 

121 correlation_id: str | None = None 

122 

123 

124@dataclass(frozen=True, kw_only=True) 

125class BulkOperationCompleted(DomainEvent): 

126 """Raised when a bulk operation completes.""" 

127 

128 resource_type: str = "" 

129 operation: str = "" 

130 resource_ids: list[Any] = field(default_factory=list) 

131 success_count: int = 0 

132 failure_count: int = 0 

133 errors: list[str] = field(default_factory=list) 

134 correlation_id: str | None = None 

135 

136 

137# ========== Operational admin events (previously events/events.py) ========== 

138 

139 

140@dataclass(frozen=True, kw_only=True) 

141class AdminEvent(DomainEvent): 

142 """Base class for all admin operational events.""" 

143 

144 correlation_id: str | None = None 

145 

146 

147@dataclass(frozen=True, kw_only=True) 

148class ResourceRestored(AdminEvent): 

149 """Event emitted when a soft-deleted resource is restored.""" 

150 

151 resource_type: str = "" 

152 resource_id: Any = None 

153 

154 

155@dataclass(frozen=True, kw_only=True) 

156class ResourceViewed(AdminEvent): 

157 """Event emitted when a resource is viewed (for audit).""" 

158 

159 resource_type: str = "" 

160 resource_id: Any = None 

161 

162 

163@dataclass(frozen=True, kw_only=True) 

164class ExportStarted(AdminEvent): 

165 """Event emitted when an export starts.""" 

166 

167 export_id: str = "" 

168 resource_type: str = "" 

169 format: str = "" 

170 total_records: int = 0 

171 

172 

173@dataclass(frozen=True, kw_only=True) 

174class ExportCompleted(AdminEvent): 

175 """Event emitted when an export completes.""" 

176 

177 export_id: str = "" 

178 resource_type: str = "" 

179 format: str = "" 

180 total_records: int = 0 

181 file_path: str = "" 

182 file_size: int = 0 

183 

184 

185@dataclass(frozen=True, kw_only=True) 

186class ExportFailed(AdminEvent): 

187 """Event emitted when an export fails.""" 

188 

189 export_id: str = "" 

190 resource_type: str = "" 

191 error: str = "" 

192 

193 

194@dataclass(frozen=True, kw_only=True) 

195class ImportStarted(AdminEvent): 

196 """Event emitted when an import starts.""" 

197 

198 import_id: str = "" 

199 resource_type: str = "" 

200 file_name: str = "" 

201 total_rows: int = 0 

202 

203 

204@dataclass(frozen=True, kw_only=True) 

205class ImportCompleted(AdminEvent): 

206 """Event emitted when an import completes.""" 

207 

208 import_id: str = "" 

209 resource_type: str = "" 

210 created_count: int = 0 

211 updated_count: int = 0 

212 skipped_count: int = 0 

213 error_count: int = 0 

214 

215 

216@dataclass(frozen=True, kw_only=True) 

217class AdminUserLoggedIn(AdminEvent): 

218 """Event emitted when an admin user logs in.""" 

219 

220 username: str = "" 

221 ip_address: str = "" 

222 user_agent: str = "" 

223 

224 

225@dataclass(frozen=True, kw_only=True) 

226class AdminUserLoggedOut(AdminEvent): 

227 """Event emitted when an admin user logs out.""" 

228 

229 username: str = "" 

230 

231 

232@dataclass(frozen=True, kw_only=True) 

233class AdminUserCreated(AdminEvent): 

234 """Event emitted when an admin user account is created.""" 

235 

236 admin_user_id: str = "" 

237 username: str = "" 

238 email: str = "" 

239 roles: list[str] = field(default_factory=list) 

240 

241 

242@dataclass(frozen=True, kw_only=True) 

243class AdminUserUpdated(AdminEvent): 

244 """Event emitted when an admin user account is updated.""" 

245 

246 admin_user_id: str = "" 

247 changes: dict[str, tuple[Any, Any]] = field(default_factory=dict) 

248 

249 

250@dataclass(frozen=True, kw_only=True) 

251class PasswordChanged(AdminEvent): 

252 """Event emitted when a password is changed.""" 

253 

254 admin_user_id: str = "" 

255 

256 

257@dataclass(frozen=True, kw_only=True) 

258class RolesAssigned(AdminEvent): 

259 """Event emitted when roles are assigned to a user.""" 

260 

261 admin_user_id: str = "" 

262 old_roles: list[str] = field(default_factory=list) 

263 new_roles: list[str] = field(default_factory=list) 

264 

265 

266@dataclass(frozen=True, kw_only=True) 

267class ActionExecuted(AdminEvent): 

268 """Event emitted when a custom action is executed.""" 

269 

270 action_name: str = "" 

271 resource_type: str | None = None 

272 resource_ids: list[Any] = field(default_factory=list) 

273 params: dict[str, Any] = field(default_factory=dict) 

274 result: dict[str, Any] = field(default_factory=dict) 

275 

276 

277@dataclass(frozen=True, kw_only=True) 

278class SettingsUpdated(AdminEvent): 

279 """Event emitted when settings are updated.""" 

280 

281 section: str = "" 

282 changes: dict[str, tuple[Any, Any]] = field(default_factory=dict) 

283 

284 

285@dataclass(frozen=True, kw_only=True) 

286class AdminStarted(AdminEvent): 

287 """Event emitted when the admin panel starts.""" 

288 

289 version: str = "" 

290 config: dict[str, Any] = field(default_factory=dict) 

291 

292 

293@dataclass(frozen=True, kw_only=True) 

294class AdminStopped(AdminEvent): 

295 """Event emitted when the admin panel stops.""" 

296 

297 

298@dataclass(frozen=True, kw_only=True) 

299class ConfigReloaded(AdminEvent): 

300 """Event emitted when configuration is reloaded.""" 

301 

302 old_config: dict[str, Any] = field(default_factory=dict) 

303 new_config: dict[str, Any] = field(default_factory=dict)