Coverage for src / lexigram / admin / cqrs / commands.py: 0%

1"""CQRS Commands for lexigram-admin. 

2 

3Defines command objects for admin operations. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from datetime import UTC, datetime 

10from typing import Any 

11from uuid import uuid4 

12 

13 

14@dataclass 

15class AdminCommand: 

16 """Base class for admin commands.""" 

17 

18 command_id: str = field(default_factory=lambda: str(uuid4())) 

19 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

20 user_id: str | None = None 

21 correlation_id: str | None = None 

22 

23 @property 

24 def name(self) -> str: 

25 return self.__class__.__name__ 

26 

27 

28# ========== Resource Commands ========== 

29 

30 

31@dataclass 

32class CreateResource(AdminCommand): 

33 """Command to create a new resource.""" 

34 

35 resource_type: str = "" 

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

37 

38 @property 

39 def name(self) -> str: 

40 return f"Create{self.resource_type.title()}" 

41 

42 

43@dataclass 

44class UpdateResource(AdminCommand): 

45 """Command to update a resource.""" 

46 

47 resource_type: str = "" 

48 resource_id: Any = None 

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

50 partial: bool = True # PATCH vs PUT 

51 

52 @property 

53 def name(self) -> str: 

54 return f"Update{self.resource_type.title()}" 

55 

56 

57@dataclass 

58class DeleteResource(AdminCommand): 

59 """Command to delete a resource.""" 

60 

61 resource_type: str = "" 

62 resource_id: Any = None 

63 soft_delete: bool = True 

64 

65 @property 

66 def name(self) -> str: 

67 return f"Delete{self.resource_type.title()}" 

68 

69 

70@dataclass 

71class RestoreResource(AdminCommand): 

72 """Command to restore a soft-deleted resource.""" 

73 

74 resource_type: str = "" 

75 resource_id: Any = None 

76 

77 

78# ========== Bulk Commands ========== 

79 

80 

81@dataclass 

82class BulkCreateResources(AdminCommand): 

83 """Command to bulk create resources.""" 

84 

85 resource_type: str = "" 

86 items: list[dict[str, Any]] = field(default_factory=list) 

87 

88 

89@dataclass 

90class BulkUpdateResources(AdminCommand): 

91 """Command to bulk update resources.""" 

92 

93 resource_type: str = "" 

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

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

96 

97 

98@dataclass 

99class BulkDeleteResources(AdminCommand): 

100 """Command to bulk delete resources.""" 

101 

102 resource_type: str = "" 

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

104 soft_delete: bool = True 

105 

106 

107# ========== Export Commands ========== 

108 

109 

110@dataclass 

111class ExportResources(AdminCommand): 

112 """Command to export resources.""" 

113 

114 resource_type: str = "" 

115 query: dict[str, Any] = field(default_factory=dict) 

116 format: str = "csv" # csv, json, xlsx 

117 columns: list[str] | None = None 

118 notify_email: str | None = None 

119 

120 

121# ========== Import Commands ========== 

122 

123 

124@dataclass 

125class ImportResources(AdminCommand): 

126 """Command to import resources from file.""" 

127 

128 resource_type: str = "" 

129 file_path: str = "" 

130 format: str = "csv" 

131 mapping: dict[str, str] | None = None # file_column -> model_field 

132 on_duplicate: str = "skip" # skip, update, error 

133 validate_only: bool = False 

134 

135 

136# ========== Action Commands ========== 

137 

138 

139@dataclass 

140class ExecuteAction(AdminCommand): 

141 """Command to execute a custom action.""" 

142 

143 action_name: str = "" 

144 resource_type: str | None = None 

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

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

147 

148 

149# ========== Settings Commands ========== 

150 

151 

152@dataclass 

153class UpdateSettings(AdminCommand): 

154 """Command to update admin settings.""" 

155 

156 settings: dict[str, Any] = field(default_factory=dict) 

157 section: str | None = None 

158 

159 

160# ========== User Management Commands ========== 

161 

162 

163@dataclass 

164class CreateAdminUser(AdminCommand): 

165 """Command to create admin user.""" 

166 

167 username: str = "" 

168 email: str = "" 

169 password: str = "" 

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

171 permissions: list[str] = field(default_factory=list) 

172 

173 

174@dataclass 

175class UpdateAdminUser(AdminCommand): 

176 """Command to update admin user.""" 

177 

178 user_id: str = "" 

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

180 

181 

182@dataclass 

183class ChangePassword(AdminCommand): 

184 """Command to change user password.""" 

185 

186 user_id: str = "" 

187 current_password: str = "" 

188 new_password: str = "" 

189 

190 

191@dataclass 

192class AssignRoles(AdminCommand): 

193 """Command to assign roles to user.""" 

194 

195 user_id: str = "" 

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

197 append: bool = False # False = replace, True = add 

198 

199 

200__all__ = [ 

201 "AdminCommand", 

202 "AssignRoles", 

203 "BulkCreateResources", 

204 "BulkDeleteResources", 

205 "BulkUpdateResources", 

206 "ChangePassword", 

207 "CreateAdminUser", 

208 "CreateResource", 

209 "DeleteResource", 

210 "ExecuteAction", 

211 "ExportResources", 

212 "ImportResources", 

213 "RestoreResource", 

214 "UpdateAdminUser", 

215 "UpdateResource", 

216 "UpdateSettings", 

217]