Coverage for src/lexigram/admin/actions/standard/export.py: 94%
70 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Export actions backed by the admin ExportService.
3Part of the ``lexigram.admin.actions.standard`` package.
4"""
6from __future__ import annotations
8from typing import TYPE_CHECKING, Any
10from lexigram.admin.actions.base import BulkAction, RowAction
11from lexigram.admin.actions.standard.utils import _extract_id, _resolve_data_source
12from lexigram.admin.actions.exceptions import ActionError
13from lexigram.admin.actions.types import (
14 ActionColor,
15 ActionContext,
16 ConfirmationConfig,
17)
18from lexigram.result import Err, Ok, Result
20if TYPE_CHECKING:
21 from lexigram.admin.services.export import ExportFormat, ExportService
22async def _resolve_export_service(ctx: ActionContext) -> ExportService | None:
23 """Resolve an ExportService from the request container.
25 Returns None when the request or container is unavailable, or when
26 resolution fails (e.g. the service is not registered).
27 """
28 if ctx.request is None:
29 return None
30 container = getattr(ctx.request.state, "container", None) or getattr(
31 ctx.request.app.state, "container", None
32 )
33 if container is None:
34 return None
35 try:
36 from lexigram.admin.services.export import ExportService
38 return await container.resolve(ExportService)
39 except Exception: # noqa: BLE001 — non-fatal
40 return None
41async def _run_export(
42 ctx: ActionContext,
43 service: ExportService,
44 data_source: Any,
45 filters: dict[str, Any],
46 *,
47 resource_name: str,
48 file_format: ExportFormat,
49 message: str,
50) -> Result[Any, Any]:
51 """Create and execute an export job, returning the job summary."""
52 from lexigram.admin.data.adapters.export_adapter import ExportDataSourceAdapter
54 job_id = service.create_job(
55 resource_name=resource_name,
56 file_format=file_format,
57 filters=filters,
58 user_id=getattr(ctx.user, "id", None),
59 )
60 result = await service.execute_export(job_id, ExportDataSourceAdapter(data_source))
61 if result.is_err():
62 return Err(result.unwrap_err())
63 job = result.unwrap()
64 return Ok(
65 {
66 "message": message,
67 "job_id": job.job_id,
68 "total_records": job.total_records,
69 "download_url": job.download_url,
70 "file_path": job.file_path,
71 }
72 )
73class ExportAction(RowAction):
74 """Export a single record through the admin ExportService."""
76 def __init__(
77 self,
78 name: str = "export",
79 label: str | None = None,
80 export_service: ExportService | None = None,
81 data_source: Any | None = None,
82 file_format: ExportFormat | None = None,
83 ) -> None:
84 super().__init__(
85 name=name,
86 label=label or "Export",
87 icon="download",
88 color=ActionColor.GRAY,
89 )
90 if file_format is None:
91 from lexigram.admin.services.export import ExportFormat
93 file_format = ExportFormat.CSV
94 self._export_service = export_service
95 self._data_source = data_source
96 self._file_format = file_format
98 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
99 record_id = self._get_record_id(record)
100 if not record_id:
101 return Err(ActionError("Cannot export a record without an id."))
102 service = self._export_service
103 if service is None:
104 service = await _resolve_export_service(ctx)
105 if service is None:
106 return Err(
107 ActionError(
108 "Export requires an ExportService; inject one via the action "
109 "constructor or register it in the request container."
110 )
111 )
112 data_source = _resolve_data_source(ctx, self._data_source)
113 if data_source is None:
114 return Err(
115 ActionError(
116 "Export requires a data source; set ctx.data_source or "
117 "ctx.metadata['data_source']."
118 )
119 )
120 return await _run_export(
121 ctx=ctx,
122 service=service,
123 data_source=data_source,
124 filters={"id": record_id},
125 resource_name=ctx.resource_name or self.name,
126 file_format=self._file_format,
127 message=f"Exported record {record_id}",
128 )
129class ExportBulkAction(BulkAction):
130 """Export multiple selected records through the admin ExportService."""
132 def __init__(
133 self,
134 name: str = "export",
135 label: str | None = None,
136 export_service: ExportService | None = None,
137 data_source: Any | None = None,
138 file_format: ExportFormat | None = None,
139 ) -> None:
140 super().__init__(
141 name=name,
142 label=label or "Export Selected",
143 icon="download",
144 color=ActionColor.GRAY,
145 )
146 if file_format is None:
147 from lexigram.admin.services.export import ExportFormat
149 file_format = ExportFormat.CSV
150 self._export_service = export_service
151 self._data_source = data_source
152 self._file_format = file_format
154 async def execute(self, records: list[Any], ctx: ActionContext) -> Result[Any, Any]:
155 record_ids = [
156 record_id
157 for record_id in (_extract_id(record) for record in records)
158 if record_id is not None
159 ]
160 if not record_ids:
161 return Err(ActionError("Cannot export records without ids."))
162 service = self._export_service
163 if service is None:
164 service = await _resolve_export_service(ctx)
165 if service is None:
166 return Err(
167 ActionError(
168 "Export requires an ExportService; inject one via the action "
169 "constructor or register it in the request container."
170 )
171 )
172 data_source = _resolve_data_source(ctx, self._data_source)
173 if data_source is None:
174 return Err(
175 ActionError(
176 "Export requires a data source; set ctx.data_source or "
177 "ctx.metadata['data_source']."
178 )
179 )
180 return await _run_export(
181 ctx=ctx,
182 service=service,
183 data_source=data_source,
184 filters={"id__in": record_ids},
185 resource_name=ctx.resource_name or self.name,
186 file_format=self._file_format,
187 message=f"Exported {len(record_ids)} record(s)",
188 )