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.contracts.events import CommandHandlerProtocol as BaseCommandHandler 

20from lexigram.contracts.events import EventBusProtocol 

21from lexigram.di.decorators import inject 

22from lexigram.logging import get_logger 

23from lexigram.result import Err, Ok, Result 

24 

25logger = get_logger(__name__) 

26 

27if TYPE_CHECKING: 

28 from lexigram.admin.cqrs.commands import ( 

29 BulkDeleteResources, 

30 CreateResource, 

31 DeleteResource, 

32 ExportResources, 

33 UpdateResource, 

34 ) 

35 from lexigram.contracts.data.data_source import DataSourceProtocol 

36 

37 

38@inject 

39class ResourceCommandHandler(BaseCommandHandler): 

40 """Handler for resource CRUD commands.""" 

41 

42 def __init__( 

43 self, 

44 data_sources: dict[str, DataSourceProtocol], 

45 event_bus: EventBusProtocol | None = None, 

46 ): 

47 self.data_sources = data_sources 

48 self.event_bus = event_bus 

49 

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

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

52 if resource_type not in self.data_sources: 

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

54 return self.data_sources[resource_type] 

55 

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

57 """Handle CreateResource command.""" 

58 try: 

59 data_source = self.get_data_source(command.resource_type) 

60 

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

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

63 

64 if self.event_bus: 

65 event = ResourceCreated( 

66 resource_type=command.resource_type, 

67 resource_id=resource_id, 

68 data=command.data, 

69 actor_id=command.user_id, 

70 correlation_id=command.correlation_id, 

71 ) 

72 await self.event_bus.publish(event) 

73 

74 logger.info( 

75 "Created %s: %s", 

76 command.resource_type, 

77 resource_id, 

78 ) 

79 

80 return Ok(result) 

81 

82 except (ValueError, ConnectionError, TimeoutError, OSError, KeyError) as e: 

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

84 return Err(str(e)) 

85 

86 async def handle_update(self, command: UpdateResource) -> Result[Any, str]: 

87 """Handle UpdateResource command.""" 

88 try: 

89 data_source = self.get_data_source(command.resource_type) 

90 

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

92 if old_item is None: 

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

94 

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

96 

97 changes = {} 

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

99 old_val = getattr(old_item, key, None) 

100 if old_val != new_val: 

101 changes[key] = (old_val, new_val) 

102 

103 if self.event_bus and changes: 

104 event = ResourceUpdated( 

105 resource_type=command.resource_type, 

106 resource_id=command.resource_id, 

107 changes=changes, 

108 actor_id=command.user_id, 

109 correlation_id=command.correlation_id, 

110 ) 

111 await self.event_bus.publish(event) 

112 

113 logger.info( 

114 "Updated %s: %s", 

115 command.resource_type, 

116 command.resource_id, 

117 ) 

118 

119 return Ok(result) 

120 

121 except (ValueError, ConnectionError, TimeoutError, OSError, KeyError) as e: 

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

123 return Err(str(e)) 

124 

125 async def handle_delete(self, command: DeleteResource) -> Result[Any, str]: 

126 """Handle DeleteResource command.""" 

127 try: 

128 data_source = self.get_data_source(command.resource_type) 

129 

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

131 

132 if not success: 

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

134 

135 if self.event_bus: 

136 event = ResourceDeleted( 

137 resource_type=command.resource_type, 

138 resource_id=command.resource_id, 

139 soft_delete=command.soft_delete, 

140 actor_id=command.user_id, 

141 correlation_id=command.correlation_id, 

142 ) 

143 await self.event_bus.publish(event) 

144 

145 logger.info( 

146 "Deleted %s: %s", 

147 command.resource_type, 

148 command.resource_id, 

149 ) 

150 

151 return Ok(None) 

152 

153 except (ValueError, ConnectionError, TimeoutError, OSError, KeyError) as e: 

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

155 return Err(str(e)) 

156 

157 async def handle_bulk_delete( 

158 self, command: BulkDeleteResources 

159 ) -> Result[Any, str]: 

160 """Handle BulkDeleteResources command.""" 

161 try: 

162 data_source = self.get_data_source(command.resource_type) 

163 

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

165 

166 if self.event_bus: 

167 event = BulkOperationCompleted( 

168 operation="delete", 

169 resource_type=command.resource_type, 

170 resource_ids=command.resource_ids, 

171 success_count=count, 

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

173 actor_id=command.user_id, 

174 correlation_id=command.correlation_id, 

175 ) 

176 await self.event_bus.publish(event) 

177 

178 logger.info( 

179 "Bulk deleted %d %s", 

180 count, 

181 command.resource_type, 

182 ) 

183 

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

185 

186 except (ValueError, ConnectionError, TimeoutError, OSError, KeyError) as e: 

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

188 return Err(str(e)) 

189 

190 

191@inject 

192class ExportCommandHandler(BaseCommandHandler): 

193 """Handler for export commands.""" 

194 

195 def __init__( 

196 self, 

197 data_sources: dict[str, DataSourceProtocol], 

198 event_bus: EventBusProtocol | None = None, 

199 export_dir: str | None = None, 

200 ): 

201 self.data_sources = data_sources 

202 self.event_bus = event_bus 

203 if export_dir: 

204 self.export_dir = export_dir 

205 else: 

206 from pathlib import Path 

207 import tempfile 

208 

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

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

211 self.export_dir = str(temp_dir) 

212 

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

214 """Handle ExportResources command.""" 

215 import csv 

216 from pathlib import Path 

217 from uuid import uuid4 

218 

219 from lexigram import serialization as json 

220 

221 export_id = str(uuid4()) 

222 

223 try: 

224 if command.resource_type not in self.data_sources: 

225 return Err( 

226 f"Unknown resource type: {command.resource_type}", 

227 ) 

228 

229 data_source = self.data_sources[command.resource_type] 

230 

231 from lexigram.admin.data.query import QuerySpec 

232 

233 qs = QuerySpec() 

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

235 qs = qs.with_where_eq(field, value) 

236 if self.event_bus: 

237 await self.event_bus.publish( 

238 ExportStarted( 

239 export_id=export_id, 

240 resource_type=command.resource_type, 

241 format=command.format, 

242 actor_id=command.user_id, 

243 ), 

244 ) 

245 

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

247 

248 import aiofiles 

249 

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

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

252 

253 if command.format == "json": 

254 items = [ 

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

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

257 ] 

258 import asyncio 

259 

260 content = await asyncio.to_thread( 

261 json.dumps_str, 

262 items, 

263 default=str, 

264 indent=2, 

265 ) 

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

267 await f.write(content) 

268 

269 elif command.format == "csv": 

270 import io 

271 

272 output = io.StringIO() 

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

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

275 fieldnames = command.columns or [ 

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

277 ] 

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

279 writer.writeheader() 

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

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

282 writer.writerow(row) 

283 

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

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

286 

287 file_size = export_path.stat().st_size 

288 

289 total_records = ( 

290 result.total 

291 if hasattr(result, "total") 

292 else len(result.items) 

293 if hasattr(result, "items") 

294 else 0 

295 ) 

296 

297 if self.event_bus: 

298 await self.event_bus.publish( 

299 ExportCompleted( 

300 export_id=export_id, 

301 resource_type=command.resource_type, 

302 format=command.format, 

303 total_records=total_records, 

304 file_path=str(export_path), 

305 file_size=file_size, 

306 actor_id=command.user_id, 

307 ), 

308 ) 

309 

310 logger.info( 

311 "Exported %d %s to %s", 

312 total_records, 

313 command.resource_type, 

314 export_path, 

315 ) 

316 

317 return Ok( 

318 { 

319 "export_id": export_id, 

320 "file_path": str(export_path), 

321 "total_records": total_records, 

322 }, 

323 ) 

324 

325 except (ValueError, ConnectionError, TimeoutError, OSError) as e: 

326 logger.exception("Export failed") 

327 return Err(str(e))