Coverage for src/lexigram/admin/handlers/admin_command_handlers.py: 0%

1"""Command handlers for lexigram-admin. 

2 

3Handlers process commands and emit corresponding events. 

4They integrate with lexigram-events CommandBusProtocol. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.admin.events import ( 

12 BulkOperationCompleted, 

13 ExportCompleted, 

14 ExportStarted, 

15 ResourceCreated, 

16 ResourceDeleted, 

17 ResourceUpdated, 

18) 

19from lexigram.admin.exceptions import AdminError, NotFoundError 

20from lexigram.contracts.events import CommandHandlerProtocol as BaseCommandHandler 

21from lexigram.contracts.events import EventBusProtocol 

22from lexigram.di.decorators import inject 

23from lexigram.logging import get_logger 

24from lexigram.result import Err, Ok, Result 

25 

26logger = get_logger(__name__) 

27 

28if TYPE_CHECKING: 

29 from lexigram.admin.cqrs.commands import ( 

30 BulkDeleteResources, 

31 CreateResource, 

32 DeleteResource, 

33 ExportResources, 

34 UpdateResource, 

35 ) 

36 from lexigram.contracts.data.data_source import DataSourceProtocol 

37 

38 

39@inject 

40class ResourceCommandHandler(BaseCommandHandler): 

41 """Handler for resource CRUD commands.""" 

42 

43 def __init__( 

44 self, 

45 data_sources: dict[str, DataSourceProtocol], 

46 event_bus: EventBusProtocol | None = None, 

47 ): 

48 self.data_sources = data_sources 

49 self.event_bus = event_bus 

50 

51 def get_data_source(self, resource_type: str) -> DataSourceProtocol: 

52 """Get data source for resource type.""" 

53 if resource_type not in self.data_sources: 

54 raise ValueError(f"Unknown resource type: {resource_type}") 

55 return self.data_sources[resource_type] 

56 

57 async def handle_create(self, command: CreateResource) -> Result[Any, AdminError]: 

58 """Handle CreateResource command.""" 

59 try: 

60 data_source = self.get_data_source(command.resource_type) 

61 

62 result = await data_source.create(command.data) 

63 resource_id = getattr(result, "id", None) 

64 

65 if self.event_bus: 

66 event = ResourceCreated( 

67 resource_type=command.resource_type, 

68 resource_id=resource_id, 

69 data=command.data, 

70 actor_id=command.user_id, 

71 correlation_id=command.correlation_id, 

72 ) 

73 await self.event_bus.publish(event) 

74 

75 logger.info( 

76 "Created %s: %s", 

77 command.resource_type, 

78 resource_id, 

79 ) 

80 

81 return Ok(result) 

82 

83 except ValueError as e: 

84 logger.exception("Failed to create %s", command.resource_type) 

85 return Err(AdminError(str(e))) 

86 

87 async def handle_update( 

88 self, command: UpdateResource 

89 ) -> Result[Any, NotFoundError | AdminError]: 

90 """Handle UpdateResource command.""" 

91 try: 

92 data_source = self.get_data_source(command.resource_type) 

93 

94 old_item = await data_source.find_one(command.resource_id) 

95 if old_item is None: 

96 return Err(NotFoundError(f"{command.resource_type} not found")) 

97 

98 result = await data_source.update(command.resource_id, command.data) 

99 

100 changes = {} 

101 for key, new_val in command.data.items(): 

102 old_val = getattr(old_item, key, None) 

103 if old_val != new_val: 

104 changes[key] = (old_val, new_val) 

105 

106 if self.event_bus and changes: 

107 event = ResourceUpdated( 

108 resource_type=command.resource_type, 

109 resource_id=command.resource_id, 

110 changes=changes, 

111 actor_id=command.user_id, 

112 correlation_id=command.correlation_id, 

113 ) 

114 await self.event_bus.publish(event) 

115 

116 logger.info( 

117 "Updated %s: %s", 

118 command.resource_type, 

119 command.resource_id, 

120 ) 

121 

122 return Ok(result) 

123 

124 except ValueError as e: 

125 logger.exception("Failed to update %s", command.resource_type) 

126 return Err(AdminError(str(e))) 

127 

128 async def handle_delete( 

129 self, command: DeleteResource 

130 ) -> Result[Any, NotFoundError | AdminError]: 

131 """Handle DeleteResource command.""" 

132 try: 

133 data_source = self.get_data_source(command.resource_type) 

134 

135 success = await data_source.delete(command.resource_id) 

136 

137 if not success: 

138 return Err(NotFoundError(f"{command.resource_type} not found")) 

139 

140 if self.event_bus: 

141 event = ResourceDeleted( 

142 resource_type=command.resource_type, 

143 resource_id=command.resource_id, 

144 soft_delete=command.soft_delete, 

145 actor_id=command.user_id, 

146 correlation_id=command.correlation_id, 

147 ) 

148 await self.event_bus.publish(event) 

149 

150 logger.info( 

151 "Deleted %s: %s", 

152 command.resource_type, 

153 command.resource_id, 

154 ) 

155 

156 return Ok(None) 

157 

158 except ValueError as e: 

159 logger.exception("Failed to delete %s", command.resource_type) 

160 return Err(AdminError(str(e))) 

161 

162 async def handle_bulk_delete( 

163 self, command: BulkDeleteResources 

164 ) -> Result[Any, AdminError]: 

165 """Handle BulkDeleteResources command.""" 

166 try: 

167 data_source = self.get_data_source(command.resource_type) 

168 

169 count = await data_source.bulk_delete(command.resource_ids) # type: ignore[arg-type] 

170 

171 if self.event_bus: 

172 event = BulkOperationCompleted( 

173 operation="delete", 

174 resource_type=command.resource_type, 

175 resource_ids=command.resource_ids, 

176 success_count=count, 

177 failure_count=len(command.resource_ids) - count, 

178 actor_id=command.user_id, 

179 correlation_id=command.correlation_id, 

180 ) 

181 await self.event_bus.publish(event) 

182 

183 logger.info( 

184 "Bulk deleted %d %s", 

185 count, 

186 command.resource_type, 

187 ) 

188 

189 return Ok({"deleted": count}) 

190 

191 except ValueError as e: 

192 logger.exception("Failed to bulk delete %s", command.resource_type) 

193 return Err(AdminError(str(e))) 

194 

195 

196@inject 

197class ExportCommandHandler(BaseCommandHandler): 

198 """Handler for export commands.""" 

199 

200 def __init__( 

201 self, 

202 data_sources: dict[str, DataSourceProtocol], 

203 event_bus: EventBusProtocol | None = None, 

204 export_dir: str | None = None, 

205 ): 

206 self.data_sources = data_sources 

207 self.event_bus = event_bus 

208 if export_dir: 

209 self.export_dir = export_dir 

210 else: 

211 from pathlib import Path 

212 import tempfile 

213 

214 temp_dir = Path(tempfile.gettempdir()) / "lexigram_exports" 

215 temp_dir.mkdir(parents=True, exist_ok=True) 

216 self.export_dir = str(temp_dir) 

217 

218 async def handle_export(self, command: ExportResources) -> Result[Any, AdminError]: 

219 """Handle ExportResources command.""" 

220 import csv 

221 from pathlib import Path 

222 from uuid import uuid4 

223 

224 from lexigram import serialization as json 

225 

226 export_id = str(uuid4()) 

227 

228 try: 

229 if command.resource_type not in self.data_sources: 

230 return Err( 

231 AdminError(f"Unknown resource type: {command.resource_type}"), 

232 ) 

233 

234 data_source = self.data_sources[command.resource_type] 

235 

236 from lexigram.admin.data.query import QuerySpec 

237 

238 qs = QuerySpec() 

239 for field, value in command.query.items(): 

240 qs = qs.with_where_eq(field, value) 

241 if self.event_bus: 

242 await self.event_bus.publish( 

243 ExportStarted( 

244 export_id=export_id, 

245 resource_type=command.resource_type, 

246 format=command.format, 

247 actor_id=command.user_id, 

248 ), 

249 ) 

250 

251 result = await data_source.find_many(qs) # type: ignore[arg-type] 

252 

253 import aiofiles 

254 

255 export_path = Path(self.export_dir) / f"{export_id}.{command.format}" 

256 export_path.parent.mkdir(parents=True, exist_ok=True) 

257 

258 if command.format == "json": 

259 items = [ 

260 {k: v for k, v in vars(item).items() if not k.startswith("_")} 

261 for item in result.items # type: ignore[attr-defined] 

262 ] 

263 import asyncio 

264 

265 content = await asyncio.to_thread( 

266 json.dumps_str, 

267 items, 

268 default=str, 

269 indent=2, 

270 ) 

271 async with aiofiles.open(export_path, "w") as f: 

272 await f.write(content) 

273 

274 elif command.format == "csv": 

275 import io 

276 

277 output = io.StringIO() 

278 if result.items: # type: ignore[attr-defined] 

279 first_item = result.items[0] # type: ignore[attr-defined] 

280 fieldnames = command.columns or [ 

281 k for k in vars(first_item) if not k.startswith("_") 

282 ] 

283 writer = csv.DictWriter(output, fieldnames=fieldnames) 

284 writer.writeheader() 

285 for item in result.items: # type: ignore[attr-defined] 

286 row = {k: getattr(item, k, "") for k in fieldnames} 

287 writer.writerow(row) 

288 

289 async with aiofiles.open(export_path, "w") as f: 

290 await f.write(output.getvalue()) 

291 

292 file_size = export_path.stat().st_size 

293 

294 total_records = ( 

295 result.total 

296 if hasattr(result, "total") 

297 else len(result.items) 

298 if hasattr(result, "items") 

299 else 0 

300 ) 

301 

302 if self.event_bus: 

303 await self.event_bus.publish( 

304 ExportCompleted( 

305 export_id=export_id, 

306 resource_type=command.resource_type, 

307 format=command.format, 

308 total_records=total_records, 

309 file_path=str(export_path), 

310 file_size=file_size, 

311 actor_id=command.user_id, 

312 ), 

313 ) 

314 

315 logger.info( 

316 "Exported %d %s to %s", 

317 total_records, 

318 command.resource_type, 

319 export_path, 

320 ) 

321 

322 return Ok( 

323 { 

324 "export_id": export_id, 

325 "file_path": str(export_path), 

326 "total_records": total_records, 

327 }, 

328 ) 

329 

330 except ValueError as e: 

331 logger.exception("Export failed") 

332 return Err(AdminError(str(e)))