Coverage for src/lexigram/admin/actions/standard/imports.py: 24%

1"""Import actions (import, import bulk) with failed-import reports. 

2 

3Part of the ``lexigram.admin.actions.standard`` package. 

4""" 

5 

6from __future__ import annotations 

7 

8import csv 

9import io 

10from typing import TYPE_CHECKING, Any 

11 

12from lexigram.admin.actions.base import BulkAction, HeaderAction 

13from lexigram.admin.actions.standard.utils import _resolve_data_source 

14from lexigram.admin.actions.exceptions import ActionError 

15from lexigram.admin.actions.types import ( 

16 ActionColor, 

17 ActionContext, 

18 ConfirmationConfig, 

19) 

20from lexigram.result import Err, Ok, Result 

21 

22if TYPE_CHECKING: 

23 from lexigram.admin.services.import_ import AdminImportService 

24async def _run_import( 

25 service: AdminImportService, 

26 content: bytes, 

27 filename: str, 

28) -> Result[Any, Any]: 

29 """Parse and commit an import, returning the result summary.""" 

30 parsed = await service.parse(content, filename) 

31 if parsed.is_err(): 

32 return Err(parsed.unwrap_err()) 

33 committed = await service.commit(parsed.unwrap()) 

34 if committed.is_err(): 

35 return Err(committed.unwrap_err()) 

36 result = committed.unwrap() 

37 payload: dict[str, Any] = { 

38 "message": f"Imported {result.created} of {result.total} record(s)", 

39 "created": result.created, 

40 "failed": result.failed, 

41 "total": result.total, 

42 } 

43 if result.failed: 

44 reports = service.reports() 

45 if reports: 

46 report = reports[-1] 

47 stem = report.source_filename.rpartition(".")[0] or "import" 

48 payload["report_id"] = report.id 

49 payload["report_filename"] = f"{stem}-import-errors.csv" 

50 return Ok(payload) 

51class _ImportReportMixin: 

52 """Shared failed-import report download helpers for import actions. 

53 

54 Depends on ``self._import_service`` (an :class:`AdminImportService` 

55 with stored reports). 

56 """ 

57 

58 _import_service: AdminImportService | None = None 

59 

60 def report_csv(self, report_id: str) -> str | None: 

61 """Return CSV content of a stored failed-import report. 

62 

63 Args: 

64 report_id: Report identifier from the import service. 

65 

66 Returns: 

67 CSV content, or None when no service is configured or the 

68 report id is unknown. 

69 """ 

70 service = self._import_service 

71 if service is None: 

72 return None 

73 report = service.get_report(report_id) 

74 if report is None: 

75 return None 

76 return report.to_csv() 

77 

78 def report_filename(self, report_id: str) -> str | None: 

79 """Derive a download filename for a stored failed-import report. 

80 

81 Args: 

82 report_id: Report identifier from the import service. 

83 

84 Returns: 

85 ``{source}-import-errors.csv`` filename, or None when no 

86 service is configured or the report id is unknown. 

87 """ 

88 service = self._import_service 

89 if service is None: 

90 return None 

91 report = service.get_report(report_id) 

92 if report is None: 

93 return None 

94 stem = report.source_filename.rpartition(".")[0] or "import" 

95 return f"{stem}-import-errors.csv" 

96class ImportAction(_ImportReportMixin, HeaderAction): 

97 """Import records into a resource through the admin import service.""" 

98 

99 def __init__( 

100 self, 

101 name: str = "import", 

102 label: str | None = None, 

103 import_service: AdminImportService | None = None, 

104 data_source: Any | None = None, 

105 file_content: bytes | None = None, 

106 filename: str | None = None, 

107 example_columns: list[str] | None = None, 

108 example_filename: str = "import-example.csv", 

109 ) -> None: 

110 super().__init__( 

111 name=name, 

112 label=label or "Import", 

113 icon="upload", 

114 color=ActionColor.GRAY, 

115 ) 

116 self._import_service = import_service 

117 self._data_source = data_source 

118 self._file_content = file_content 

119 self._filename = filename 

120 self._example_columns = example_columns or [] 

121 self._example_filename = example_filename 

122 

123 def example_csv(self) -> str: 

124 """Build a header-only example CSV from ``example_columns``. 

125 

126 Mirrors Filament's ``ImportAction::exampleCsv()``. Returns an empty 

127 string when no example columns are configured. 

128 """ 

129 if not self._example_columns: 

130 return "" 

131 buffer = io.StringIO() 

132 writer = csv.writer(buffer, lineterminator="\n") 

133 writer.writerow(self._example_columns) 

134 return buffer.getvalue() 

135 

136 @property 

137 def example_filename(self) -> str: 

138 """Download filename for the example CSV template.""" 

139 return self._example_filename 

140 

141 async def execute(self, record: None, ctx: ActionContext) -> Result[Any, Any]: 

142 content = self._file_content or ctx.metadata.get("file_content") 

143 if content is None: 

144 return Err( 

145 ActionError( 

146 "Import requires file content; pass file_content to the action " 

147 "or set ctx.metadata['file_content']." 

148 ) 

149 ) 

150 filename = self._filename or ctx.metadata.get("filename") or "import.csv" 

151 service = self._import_service 

152 if service is None: 

153 data_source = _resolve_data_source(ctx, self._data_source) 

154 if data_source is None: 

155 return Err( 

156 ActionError( 

157 "Import requires an AdminImportService or a data source; " 

158 "inject one or set ctx.data_source." 

159 ) 

160 ) 

161 from lexigram.admin.services.import_ import AdminImportService 

162 

163 service = AdminImportService(data_source=data_source) 

164 return await _run_import(service, content, filename) 

165class ImportBulkAction(_ImportReportMixin, BulkAction): 

166 """Import multiple records through the admin import service.""" 

167 

168 def __init__( 

169 self, 

170 name: str = "import", 

171 label: str | None = None, 

172 import_service: AdminImportService | None = None, 

173 data_source: Any | None = None, 

174 file_content: bytes | None = None, 

175 filename: str | None = None, 

176 ) -> None: 

177 super().__init__( 

178 name=name, 

179 label=label or "Import Selected", 

180 icon="upload", 

181 color=ActionColor.GRAY, 

182 ) 

183 self._import_service = import_service 

184 self._data_source = data_source 

185 self._file_content = file_content 

186 self._filename = filename 

187 

188 async def execute(self, records: list[Any], ctx: ActionContext) -> Result[Any, Any]: 

189 content = self._file_content or ctx.metadata.get("file_content") 

190 if content is None: 

191 return Err( 

192 ActionError( 

193 "Import requires file content; pass file_content to the action " 

194 "or set ctx.metadata['file_content']." 

195 ) 

196 ) 

197 filename = self._filename or ctx.metadata.get("filename") or "import.csv" 

198 service = self._import_service 

199 if service is None: 

200 data_source = _resolve_data_source(ctx, self._data_source) 

201 if data_source is None: 

202 return Err( 

203 ActionError( 

204 "Import requires an AdminImportService or a data source; " 

205 "inject one or set ctx.data_source." 

206 ) 

207 ) 

208 from lexigram.admin.services.import_ import AdminImportService 

209 

210 service = AdminImportService(data_source=data_source) 

211 return await _run_import(service, content, filename)