Coverage for src/lexigram/admin/resources/archive_ops.py: 24%

1"""Archive lifecycle operations for Admin Resources. 

2 

3Clone/duplicate, soft-delete restore, and hard purge flows with their 

4``before_*`` / ``after_*`` extension hooks. Composed into 

5:class:`~lexigram.admin.resources.base.Resource` via inheritance so the 

6methods remain part of every resource's public surface: 

7 

8 class MyResource(ArchiveOperationsMixin): ... 

9 

10Subclasses override the hook pairs to customise behaviour; the orchestrators 

11(``duplicate`` / ``restore`` / ``purge``) handle data-source plumbing. 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import TYPE_CHECKING, Any 

17 

18if TYPE_CHECKING: 

19 from lexigram.admin.data.data_source import IDataSource 

20 

21 

22class ArchiveOperationsMixin: 

23 """Clone / restore / purge operations plus their lifecycle hooks. 

24 

25 Requires the composing class to provide an attached ``_data_source`` 

26 (set by :class:`~lexigram.admin.resources.base.Resource.__init__`). 

27 """ 

28 

29 _data_source: IDataSource | None 

30 

31 async def before_clone(self, data: dict) -> dict: 

32 """Hook called before a record is cloned. 

33 

34 Strips the ``id`` field (so a new ID is assigned) and 

35 appends `` (Copy)`` to the ``name`` field. Override 

36 to customise clone behaviour. 

37 

38 Args: 

39 data: Record data dict fetched from the data source. 

40 

41 Returns: 

42 Modified data dict to be passed to ``create``. 

43 """ 

44 data.pop("id", None) 

45 if "name" in data: 

46 data["name"] = f"{data['name']} (Copy)" 

47 return data 

48 

49 async def after_clone(self, record: Any) -> None: 

50 """Hook called after a record has been cloned. 

51 

52 Args: 

53 record: The newly created record returned by the data source. 

54 """ 

55 

56 async def duplicate(self, item_id: Any) -> Any: 

57 """Duplicate (clone) a record by its identifier. 

58 

59 Fetches the existing record via the attached data source, 

60 calls :meth:`before_clone` to prepare the data, creates 

61 a new record, and calls :meth:`after_clone` with the result. 

62 

63 Args: 

64 item_id: Identifier of the record to clone. 

65 

66 Returns: 

67 The newly created record. 

68 

69 Raises: 

70 RuntimeError: If no data source is attached. 

71 """ 

72 if self._data_source is None: 

73 raise RuntimeError("No data source attached to this resource") 

74 

75 original = await self._data_source.find_one(item_id) 

76 data: dict = dict(original) if isinstance(original, dict) else {} 

77 if not data and hasattr(original, "__dict__"): 

78 data = dict(original.__dict__) 

79 data = await self.before_clone(data) 

80 new_record = await self._data_source.create(data) 

81 await self.after_clone(new_record) 

82 return new_record 

83 

84 async def before_restore(self, data: dict) -> dict: 

85 """Hook called before a soft-deleted record is restored. 

86 

87 Sets ``deleted_at`` to ``None`` by default. Override to 

88 customise restore behaviour. 

89 

90 Args: 

91 data: Record data dict fetched from the data source. 

92 

93 Returns: 

94 Modified data dict to be passed to ``update``. 

95 """ 

96 return {"deleted_at": None} 

97 

98 async def after_restore(self, record: Any) -> None: 

99 """Hook called after a record has been restored. 

100 

101 Args: 

102 record: The restored record returned by the data source. 

103 """ 

104 

105 async def restore(self, item_id: Any) -> Any: 

106 """Restore a soft-deleted record. 

107 

108 Fetches the existing record, calls :meth:`before_restore` to 

109 prepare the data, updates the record via the data source, and 

110 calls :meth:`after_restore` with the result. 

111 

112 Args: 

113 item_id: Identifier of the record to restore. 

114 

115 Returns: 

116 The restored record. 

117 

118 Raises: 

119 RuntimeError: If no data source is attached. 

120 """ 

121 if self._data_source is None: 

122 raise RuntimeError("No data source attached to this resource") 

123 

124 original = await self._data_source.find_one(item_id) 

125 data: dict = dict(original) if isinstance(original, dict) else {} 

126 if not data and hasattr(original, "__dict__"): 

127 data = dict(original.__dict__) 

128 data = await self.before_restore(data) 

129 new_record = await self._data_source.update(item_id, data) 

130 await self.after_restore(new_record) 

131 return new_record 

132 

133 async def before_purge(self, data: dict) -> dict: 

134 """Hook called before a record is permanently purged. 

135 

136 Args: 

137 data: Record data dict fetched from the data source. 

138 

139 Returns: 

140 Modified data dict (default: unchanged). 

141 """ 

142 return data 

143 

144 async def after_purge(self, item_id: Any) -> None: 

145 """Hook called after a record has been permanently purged. 

146 

147 Args: 

148 item_id: Identifier of the purged record. 

149 """ 

150 

151 async def purge(self, item_id: Any) -> None: 

152 """Permanently delete (purge) a record. 

153 

154 Fetches the existing record, calls :meth:`before_purge` to 

155 prepare the data, hard-deletes via the data source, and calls 

156 :meth:`after_purge` with the item id. 

157 

158 Args: 

159 item_id: Identifier of the record to purge. 

160 

161 Raises: 

162 RuntimeError: If no data source is attached. 

163 """ 

164 if self._data_source is None: 

165 raise RuntimeError("No data source attached to this resource") 

166 

167 original = await self._data_source.find_one(item_id) 

168 data: dict = dict(original) if isinstance(original, dict) else {} 

169 if not data and hasattr(original, "__dict__"): 

170 data = dict(original.__dict__) 

171 await self.before_purge(data) 

172 await self._data_source.delete(item_id) 

173 await self.after_purge(item_id) 

174 

175 

176__all__ = ["ArchiveOperationsMixin"]